From c0e949b12fdcdcb40f65d6898bc0c41689f0ff3f Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Tue, 1 Mar 2022 11:25:10 +0000 Subject: Use wait_until and a deadline instead of a timeout. This reduces the need to smaple the time significantly. --- .../vespa/storage/persistence/filestorage/filestorhandler.h | 12 ++++++++++-- .../storage/persistence/filestorage/filestorhandlerimpl.cpp | 12 +++++------- .../storage/persistence/filestorage/filestorhandlerimpl.h | 9 ++------- storage/src/vespa/storage/persistence/persistencethread.cpp | 7 +++++-- 4 files changed, 22 insertions(+), 18 deletions(-) (limited to 'storage') diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h b/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h index 5c243ea4af9..250bbe369c9 100644 --- a/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h +++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h @@ -128,6 +128,7 @@ public: CLOSED }; + FileStorHandler() : _getNextMessageTimout(100ms) { } virtual ~FileStorHandler() = default; @@ -170,7 +171,12 @@ public: * * @param stripe The stripe to get messages for */ - virtual LockedMessage getNextMessage(uint32_t stripeId) = 0; + virtual LockedMessage getNextMessage(uint32_t stripeId, vespalib::steady_time timeout_end) = 0; + + /** Only used for testing, should be removed */ + LockedMessage getNextMessage(uint32_t stripeId) { + return getNextMessage(stripeId, vespalib::steady_clock::now() + _getNextMessageTimout); + } /** * Lock a bucket. By default, each file stor thread has the locks of all @@ -268,7 +274,7 @@ public: virtual uint32_t getQueueSize() const = 0; // Commands used by testing - virtual void setGetNextMessageTimeout(vespalib::duration timeout) = 0; + void setGetNextMessageTimeout(vespalib::duration timeout) { _getNextMessageTimout = timeout; } virtual std::string dumpQueue() const = 0; @@ -277,6 +283,8 @@ public: virtual vespalib::SharedOperationThrottler& operation_throttler() const noexcept = 0; virtual void set_throttle_apply_bucket_diff_ops(bool throttle_apply_bucket_diff) noexcept = 0; +private: + vespalib::duration _getNextMessageTimout; }; } // storage diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp index b5de5a233cc..77617cecad3 100644 --- a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp +++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp @@ -5,7 +5,6 @@ #include "mergestatus.h" #include #include -#include #include #include #include @@ -55,7 +54,6 @@ FileStorHandlerImpl::FileStorHandlerImpl(uint32_t numThreads, uint32_t numStripe _stripes(), _messageSender(sender), _bucketIdFactory(_component.getBucketIdFactory()), - _getNextMessageTimeout(100ms), _max_active_merges_per_stripe(per_stripe_merge_limit(numThreads, numStripes)), _paused(false), _throttle_apply_bucket_diff_ops(false), @@ -377,13 +375,13 @@ FileStorHandlerImpl::makeQueueTimeoutReply(api::StorageMessage& msg) } FileStorHandler::LockedMessage -FileStorHandlerImpl::getNextMessage(uint32_t stripeId) +FileStorHandlerImpl::getNextMessage(uint32_t stripeId, vespalib::steady_time timeout_end) { if (!tryHandlePause()) { return {}; // Still paused, return to allow tick. } - return getNextMessage(stripeId, _getNextMessageTimeout); + return _stripes[stripeId].getNextMessage(timeout_end); } std::shared_ptr @@ -919,7 +917,7 @@ FileStorHandlerImpl::Stripe::operation_type_should_be_throttled(api::MessageType } FileStorHandler::LockedMessage -FileStorHandlerImpl::Stripe::getNextMessage(vespalib::duration timeout) +FileStorHandlerImpl::Stripe::getNextMessage(vespalib::steady_time timeout_end) { std::unique_lock guard(*_lock); ThrottleToken throttle_token; @@ -955,12 +953,12 @@ FileStorHandlerImpl::Stripe::getNextMessage(vespalib::duration timeout) // Depending on whether we were blocked due to no usable ops in queue or throttling, // wait for either the queue or throttler to (hopefully) have some fresh stuff for us. if (!was_throttled) { - _cond->wait_for(guard, timeout); + _cond->wait_until(guard, timeout_end); } else { // Have to release lock before doing a blocking throttle token fetch, since it // prevents RPC threads from pushing onto the queue. guard.unlock(); - throttle_token = _owner.operation_throttler().blocking_acquire_one(timeout); + throttle_token = _owner.operation_throttler().blocking_acquire_one(timeout_end); guard.lock(); if (!throttle_token.valid()) { _metrics->timeouts_waiting_for_throttle_token.inc(); diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h index 1bc0ab87b1c..c18b51c5d10 100644 --- a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h +++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h @@ -132,7 +132,7 @@ public: std::shared_ptr lock(const document::Bucket & bucket, api::LockingRequirements lockReq); void failOperations(const document::Bucket & bucket, const api::ReturnCode & code); - FileStorHandler::LockedMessage getNextMessage(vespalib::duration timeout); + FileStorHandler::LockedMessage getNextMessage(vespalib::steady_time timeout_end); void dumpQueue(std::ostream & os) const; void dumpActiveHtml(std::ostream & os) const; void dumpQueueHtml(std::ostream & os) const; @@ -195,7 +195,6 @@ public: ServiceLayerComponentRegister&, std::unique_ptr); ~FileStorHandlerImpl() override; - void setGetNextMessageTimeout(vespalib::duration timeout) override { _getNextMessageTimeout = timeout; } void flush(bool killPendingMerges) override; void setDiskState(DiskState state) override; @@ -204,7 +203,7 @@ public: bool schedule(const std::shared_ptr&) override; ScheduleAsyncResult schedule_and_get_next_async_message(const std::shared_ptr& msg) override; - FileStorHandler::LockedMessage getNextMessage(uint32_t stripeId) override; + FileStorHandler::LockedMessage getNextMessage(uint32_t stripeId, vespalib::steady_time timeout_end) override; void remapQueueAfterJoin(const RemapInfo& source, RemapInfo& target) override; void remapQueueAfterSplit(const RemapInfo& source, RemapInfo& target1, RemapInfo& target2) override; @@ -270,7 +269,6 @@ private: const document::BucketIdFactory& _bucketIdFactory; mutable std::mutex _mergeStatesLock; std::map> _mergeStates; - vespalib::duration _getNextMessageTimeout; const uint32_t _max_active_merges_per_stripe; // Read concurrently by stripes. mutable std::mutex _pauseMonitor; mutable std::condition_variable _pauseCond; @@ -355,9 +353,6 @@ private: Stripe & stripe(const document::Bucket & bucket) { return _stripes[stripe_index(bucket)]; } - FileStorHandler::LockedMessage getNextMessage(uint32_t stripeId, vespalib::duration timeout) { - return _stripes[stripeId].getNextMessage(timeout); - } ActiveOperationsStats get_active_operations_stats(bool reset_min_max) const override; }; diff --git a/storage/src/vespa/storage/persistence/persistencethread.cpp b/storage/src/vespa/storage/persistence/persistencethread.cpp index b89c60d4720..208481bde27 100644 --- a/storage/src/vespa/storage/persistence/persistencethread.cpp +++ b/storage/src/vespa/storage/persistence/persistencethread.cpp @@ -33,10 +33,13 @@ PersistenceThread::run(framework::ThreadHandle& thread) { LOG(debug, "Started persistence thread"); + vespalib::duration max_wait_time = vespalib::adjustTimeoutByDetectedHz(100ms); while (!thread.interrupted()) { - thread.registerTick(); + vespalib::steady_time now = vespalib::steady_clock::now(); + thread.registerTick(framework::UNKNOWN_CYCLE, now); - FileStorHandler::LockedMessage lock(_fileStorHandler.getNextMessage(_stripeId)); + vespalib::steady_time doom = now + max_wait_time; + FileStorHandler::LockedMessage lock(_fileStorHandler.getNextMessage(_stripeId, doom)); if (lock.lock) { _persistenceHandler.processLockedMessage(std::move(lock)); -- cgit v1.2.3