From 509d5d5bedab5c2b52df7b4e6458ef2a1cd607b3 Mon Sep 17 00:00:00 2001 From: Haley Connelly Date: Fri, 23 Sep 2022 09:39:18 +0000 Subject: SERVER-69894 Combine PriorityTicketHolder and SchedulingTicketHolder --- src/mongo/util/concurrency/ticketholder.cpp | 106 ++++++++++++---------------- src/mongo/util/concurrency/ticketholder.h | 84 +++++++++------------- 2 files changed, 80 insertions(+), 110 deletions(-) (limited to 'src') diff --git a/src/mongo/util/concurrency/ticketholder.cpp b/src/mongo/util/concurrency/ticketholder.cpp index d910f292129..554a5f1c888 100644 --- a/src/mongo/util/concurrency/ticketholder.cpp +++ b/src/mongo/util/concurrency/ticketholder.cpp @@ -67,6 +67,28 @@ void updateQueueStatsOnTicketAcquisition(ServiceContext* serviceContext, queueStats.totalStartedProcessing.fetchAndAddRelaxed(1); } +/** + * Appends the standard statistics stored in QueueStats to BSONObjBuilder b; + */ +void appendCommonQueueImplStats(BSONObjBuilder& b, + const TicketHolderWithQueueingStats::QueueStats& stats) { + auto removed = stats.totalRemovedQueue.loadRelaxed(); + auto added = stats.totalAddedQueue.loadRelaxed(); + + b.append("addedToQueue", added); + b.append("removedFromQueue", removed); + b.append("queueLength", std::max(static_cast(added - removed), 0)); + + auto finished = stats.totalFinishedProcessing.loadRelaxed(); + auto started = stats.totalStartedProcessing.loadRelaxed(); + b.append("startedProcessing", started); + b.append("processing", std::max(static_cast(started - finished), 0)); + b.append("finishedProcessing", finished); + b.append("totalTimeProcessingMicros", stats.totalTimeProcessingMicros.loadRelaxed()); + b.append("canceled", stats.totalCanceled.loadRelaxed()); + b.append("newAdmissions", stats.totalNewAdmissions.loadRelaxed()); + b.append("totalTimeQueuedMicros", stats.totalTimeQueuedMicros.loadRelaxed()); +} } // namespace TicketHolder* TicketHolder::get(ServiceContext* svcCtx) { @@ -283,20 +305,7 @@ boost::optional TicketHolderWithQueueingStats::waitForTicketUntil(Operat } void SemaphoreTicketHolder::_appendImplStats(BSONObjBuilder& b) const { - auto removed = _semaphoreStats.totalRemovedQueue.loadRelaxed(); - auto added = _semaphoreStats.totalAddedQueue.loadRelaxed(); - b.append("addedToQueue", added); - b.append("removedFromQueue", removed); - b.append("queueLength", std::max(static_cast(added - removed), 0)); - auto finished = _semaphoreStats.totalFinishedProcessing.loadRelaxed(); - auto started = _semaphoreStats.totalStartedProcessing.loadRelaxed(); - b.append("startedProcessing", started); - b.append("processing", std::max(static_cast(started - finished), 0)); - b.append("finishedProcessing", finished); - b.append("totalTimeProcessingMicros", _semaphoreStats.totalTimeProcessingMicros.loadRelaxed()); - b.append("canceled", _semaphoreStats.totalCanceled.loadRelaxed()); - b.append("newAdmissions", _semaphoreStats.totalNewAdmissions.loadRelaxed()); - b.append("totalTimeQueuedMicros", _semaphoreStats.totalTimeQueuedMicros.loadRelaxed()); + appendCommonQueueImplStats(b, _semaphoreStats); } #if defined(__linux__) namespace { @@ -486,11 +495,9 @@ void SemaphoreTicketHolder::_resize(int newSize, int oldSize) noexcept { } #endif -SchedulingTicketHolder::SchedulingTicketHolder(int numTickets, - unsigned int numQueues, - ServiceContext* serviceContext) +PriorityTicketHolder::PriorityTicketHolder(int numTickets, ServiceContext* serviceContext) : TicketHolderWithQueueingStats(numTickets, serviceContext), _serviceContext(serviceContext) { - for (std::size_t i = 0; i < numQueues; i++) { + for (std::size_t i = 0; i < static_cast(QueueType::QueueTypeSize); i++) { _queues.emplace_back(this); } _queues.shrink_to_fit(); @@ -498,17 +505,17 @@ SchedulingTicketHolder::SchedulingTicketHolder(int numTickets, _enqueuedElements.store(0); } -SchedulingTicketHolder::~SchedulingTicketHolder() {} +PriorityTicketHolder::~PriorityTicketHolder() {} -int SchedulingTicketHolder::available() const { +int PriorityTicketHolder::available() const { return _ticketsAvailable.load(); } -int SchedulingTicketHolder::queued() const { +int PriorityTicketHolder::queued() const { return _enqueuedElements.loadRelaxed(); } -void SchedulingTicketHolder::_releaseToTicketPoolImpl(AdmissionContext* admCtx) noexcept { +void PriorityTicketHolder::_releaseToTicketPoolImpl(AdmissionContext* admCtx) noexcept { // Tickets acquired with priority kImmediate are not generated from the pool of available // tickets, and thus should never be returned to the pool of available tickets. invariant(admCtx && admCtx->getPriority() != AdmissionContext::Priority::kImmediate); @@ -533,7 +540,7 @@ void SchedulingTicketHolder::_releaseToTicketPoolImpl(AdmissionContext* admCtx) _dequeueWaitingThread(); } -bool SchedulingTicketHolder::_tryAcquireTicket() { +bool PriorityTicketHolder::_tryAcquireTicket() { auto remaining = _ticketsAvailable.subtractAndFetch(1); if (remaining < 0) { _ticketsAvailable.addAndFetch(1); @@ -542,7 +549,7 @@ bool SchedulingTicketHolder::_tryAcquireTicket() { return true; } -boost::optional SchedulingTicketHolder::_tryAcquireImpl(AdmissionContext* admCtx) { +boost::optional PriorityTicketHolder::_tryAcquireImpl(AdmissionContext* admCtx) { invariant(admCtx); auto hasAcquired = _tryAcquireTicket(); @@ -552,10 +559,10 @@ boost::optional SchedulingTicketHolder::_tryAcquireImpl(AdmissionContext return boost::none; } -boost::optional SchedulingTicketHolder::_waitForTicketUntilImpl(OperationContext* opCtx, - AdmissionContext* admCtx, - Date_t until, - WaitMode waitMode) { +boost::optional PriorityTicketHolder::_waitForTicketUntilImpl(OperationContext* opCtx, + AdmissionContext* admCtx, + Date_t until, + WaitMode waitMode) { invariant(admCtx); auto& queue = _getQueueToUse(admCtx); @@ -574,7 +581,7 @@ boost::optional SchedulingTicketHolder::_waitForTicketUntilImpl(Operatio } } -void SchedulingTicketHolder::_resize(int newSize, int oldSize) noexcept { +void PriorityTicketHolder::_resize(int newSize, int oldSize) noexcept { auto difference = newSize - oldSize; _ticketsAvailable.fetchAndAdd(difference); @@ -592,7 +599,7 @@ void SchedulingTicketHolder::_resize(int newSize, int oldSize) noexcept { // have to wait until the current ticket holders release their tickets. } -bool SchedulingTicketHolder::Queue::attemptToDequeue() { +bool PriorityTicketHolder::Queue::attemptToDequeue() { auto threadsToBeWoken = _threadsToBeWoken.load(); while (threadsToBeWoken < _queuedThreads) { auto canDequeue = _threadsToBeWoken.compareAndSwap(&threadsToBeWoken, threadsToBeWoken + 1); @@ -604,7 +611,7 @@ bool SchedulingTicketHolder::Queue::attemptToDequeue() { return false; } -void SchedulingTicketHolder::Queue::_signalThreadWoken() { +void PriorityTicketHolder::Queue::_signalThreadWoken() { auto currentThreadsToBeWoken = _threadsToBeWoken.load(); while (currentThreadsToBeWoken > 0) { if (_threadsToBeWoken.compareAndSwap(¤tThreadsToBeWoken, @@ -614,10 +621,10 @@ void SchedulingTicketHolder::Queue::_signalThreadWoken() { } } -bool SchedulingTicketHolder::Queue::enqueue(OperationContext* opCtx, - EnqueuerLockGuard& queueLock, - const Date_t& until, - WaitMode waitMode) { +bool PriorityTicketHolder::Queue::enqueue(OperationContext* opCtx, + EnqueuerLockGuard& queueLock, + const Date_t& until, + WaitMode waitMode) { _queuedThreads++; // Before exiting we remove ourselves from the count of queued threads, we are still holding the // lock here so this is safe. @@ -653,9 +660,6 @@ bool SchedulingTicketHolder::Queue::enqueue(OperationContext* opCtx, return true; } -PriorityTicketHolder::PriorityTicketHolder(int numTickets, ServiceContext* serviceContext) - : SchedulingTicketHolder(numTickets, 3, serviceContext) {} - void PriorityTicketHolder::_dequeueWaitingThread() { // There should never be anything to dequeue from 'QueueType::ImmediatePriorityNoOpQueue' since // 'kImmediate' operations should always bypass the need to queue. @@ -673,14 +677,14 @@ void PriorityTicketHolder::_appendImplStats(BSONObjBuilder& b) const { BSONObjBuilder bbb(b.subobjStart("lowPriority")); auto& lowPriorityTicketStats = _queues[static_cast(QueueType::LowPriorityQueue)].getStats(); - _appendPriorityStats(bbb, lowPriorityTicketStats); + appendCommonQueueImplStats(bbb, lowPriorityTicketStats); bbb.done(); } { BSONObjBuilder bbb(b.subobjStart("normalPriority")); auto& normalPriorityTicketStats = _queues[static_cast(QueueType::NormalPriorityQueue)].getStats(); - _appendPriorityStats(bbb, normalPriorityTicketStats); + appendCommonQueueImplStats(bbb, normalPriorityTicketStats); bbb.done(); } { @@ -702,27 +706,7 @@ void PriorityTicketHolder::_appendImplStats(BSONObjBuilder& b) const { } } -void PriorityTicketHolder::_appendPriorityStats(BSONObjBuilder& b, const QueueStats& stats) const { - auto removed = stats.totalRemovedQueue.loadRelaxed(); - auto added = stats.totalAddedQueue.loadRelaxed(); - - b.append("addedToQueue", added); - b.append("removedFromQueue", removed); - b.append("queueLength", std::max(static_cast(added - removed), 0)); - - auto finished = stats.totalFinishedProcessing.loadRelaxed(); - auto started = stats.totalStartedProcessing.loadRelaxed(); - b.append("startedProcessing", started); - b.append("processing", std::max(static_cast(started - finished), 0)); - b.append("finishedProcessing", finished); - b.append("totalTimeProcessingMicros", stats.totalTimeProcessingMicros.loadRelaxed()); - b.append("canceled", stats.totalCanceled.loadRelaxed()); - b.append("newAdmissions", stats.totalNewAdmissions.loadRelaxed()); - b.append("totalTimeQueuedMicros", stats.totalTimeQueuedMicros.loadRelaxed()); -} - -SchedulingTicketHolder::Queue& PriorityTicketHolder::_getQueueToUse( - const AdmissionContext* admCtx) noexcept { +PriorityTicketHolder::Queue& PriorityTicketHolder::_getQueueToUse(const AdmissionContext* admCtx) { auto priority = admCtx->getPriority(); switch (priority) { case AdmissionContext::Priority::kLow: diff --git a/src/mongo/util/concurrency/ticketholder.h b/src/mongo/util/concurrency/ticketholder.h index 60f5d167769..65cfddb753a 100644 --- a/src/mongo/util/concurrency/ticketholder.h +++ b/src/mongo/util/concurrency/ticketholder.h @@ -324,7 +324,21 @@ private: * Waiters will get placed in a specific internal queue according to some logic. * Releasers will wake up a waiter from a group chosen according to some logic. */ -class SchedulingTicketHolder : public TicketHolderWithQueueingStats { +class PriorityTicketHolder : public TicketHolderWithQueueingStats { +protected: +public: + explicit PriorityTicketHolder(int numTickets, ServiceContext* serviceContext); + ~PriorityTicketHolder() override; + + int available() const override final; + + int queued() const override final; + + bool recordImmediateTicketStatistics() noexcept override final { + return true; + }; + +private: // Using a shared_mutex is fine here because usual considerations for avoiding them do not apply // in this case: // * Operations are short and do not block while holding the lock (i.e. they only do CPU-bound @@ -338,10 +352,10 @@ class SchedulingTicketHolder : public TicketHolderWithQueueingStats { using QueueMutex = std::shared_mutex; // NOLINT using ReleaserLockGuard = std::shared_lock; // NOLINT using EnqueuerLockGuard = std::unique_lock; // NOLINT -protected: + class Queue { public: - Queue(SchedulingTicketHolder* holder) : _holder(holder){}; + Queue(PriorityTicketHolder* holder) : _holder(holder){}; Queue(Queue&& other) : _queuedThreads(other._queuedThreads), @@ -378,28 +392,19 @@ protected: int _queuedThreads{0}; AtomicWord _threadsToBeWoken{0}; stdx::condition_variable _cv; - SchedulingTicketHolder* _holder; + PriorityTicketHolder* _holder; QueueStats _stats; }; - std::vector _queues; - -public: - explicit SchedulingTicketHolder(int numTickets, - unsigned int numQueues, - ServiceContext* serviceContext); - ~SchedulingTicketHolder() override; - - int available() const override final; - - int queued() const override final; - - bool recordImmediateTicketStatistics() noexcept override final { - return true; + enum class QueueType : unsigned int { + LowPriorityQueue = 0, + NormalPriorityQueue = 1, + // Exclusively used for statistics tracking. This queue should never have any processes + // 'queued'. + ImmediatePriorityNoOpQueue = 2, + QueueTypeSize = 3 }; -private: - bool _tryAcquireTicket(); boost::optional _tryAcquireImpl(AdmissionContext* admCtx) override final; @@ -412,7 +417,11 @@ private: void _resize(int newSize, int oldSize) noexcept override final; - QueueStats& _getQueueStatsToUse(const AdmissionContext* admCtx) noexcept override = 0; + QueueStats& _getQueueStatsToUse(const AdmissionContext* admCtx) noexcept override final; + + void _appendImplStats(BSONObjBuilder& b) const override final; + + bool _tryAcquireTicket(); /** * Wakes up a waiting thread (if it exists) in order for it to attempt to obtain a ticket. @@ -426,14 +435,14 @@ private: * - The number of items in each queue will not change during the execution * - No other thread will proceed to wait during the execution of the method */ - virtual void _dequeueWaitingThread() = 0; - - void _appendImplStats(BSONObjBuilder& b) const override = 0; + void _dequeueWaitingThread(); /** * Selects the queue to use for the current thread given the provided arguments. */ - virtual Queue& _getQueueToUse(const AdmissionContext* admCtx) noexcept = 0; + Queue& _getQueueToUse(const AdmissionContext* admCtx); + + std::vector _queues; QueueMutex _queueMutex; AtomicWord _ticketsAvailable; @@ -441,29 +450,6 @@ private: ServiceContext* _serviceContext; }; -class PriorityTicketHolder final : public SchedulingTicketHolder { -public: - explicit PriorityTicketHolder(int numTickets, ServiceContext* serviceContext); - -private: - enum class QueueType : unsigned int { - LowPriorityQueue = 0, - NormalPriorityQueue = 1, - // Exclusively used for statistics tracking. This queue should never have any processes - // 'queued'. - ImmediatePriorityNoOpQueue = 2, - QueueTypeSize = 3 - }; - - void _dequeueWaitingThread() override final; - QueueStats& _getQueueStatsToUse(const AdmissionContext* admCtx) noexcept override final; - - void _appendImplStats(BSONObjBuilder& b) const override final; - void _appendPriorityStats(BSONObjBuilder& b, const QueueStats& stats) const; - - Queue& _getQueueToUse(const AdmissionContext* admCtx) noexcept override final; -}; - /** * RAII-style movable token that gets generated when a ticket is acquired and is automatically * released when going out of scope. @@ -473,7 +459,7 @@ class Ticket { friend class ReaderWriterTicketHolder; friend class TicketHolderWithQueueingStats; friend class SemaphoreTicketHolder; - friend class SchedulingTicketHolder; + friend class PriorityTicketHolder; public: Ticket(Ticket&& t) : _ticketholder(t._ticketholder), _admissionContext(t._admissionContext) { -- cgit v1.2.1