aboutsummaryrefslogtreecommitdiffstats
path: root/staging_vespalib
diff options
context:
space:
mode:
authorHenning Baldersheim <balder@verizonmedia.com>2020-04-04 15:49:47 +0000
committerHenning Baldersheim <balder@verizonmedia.com>2020-04-04 15:49:47 +0000
commit607c7cb232f436c2f094f1cb8cde03e76b6e360d (patch)
tree4f2500ebb05519891c3b7cd1eb850712cb2d91f3 /staging_vespalib
parentbc1ca93c91c88630f4b61e98169b224143b8829c (diff)
Control nap time
Diffstat (limited to 'staging_vespalib')
-rw-r--r--staging_vespalib/src/vespa/vespalib/util/singleexecutor.cpp13
-rw-r--r--staging_vespalib/src/vespa/vespalib/util/singleexecutor.h3
2 files changed, 13 insertions, 3 deletions
diff --git a/staging_vespalib/src/vespa/vespalib/util/singleexecutor.cpp b/staging_vespalib/src/vespa/vespalib/util/singleexecutor.cpp
index 90eb18c23ef..6492c301fe5 100644
--- a/staging_vespalib/src/vespa/vespalib/util/singleexecutor.cpp
+++ b/staging_vespalib/src/vespa/vespalib/util/singleexecutor.cpp
@@ -6,6 +6,10 @@
namespace vespalib {
SingleExecutor::SingleExecutor(uint32_t taskLimit)
+ : SingleExecutor(taskLimit, taskLimit/10, 5ms)
+{ }
+
+SingleExecutor::SingleExecutor(uint32_t taskLimit, uint32_t watermark, duration napTime)
: _taskLimit(vespalib::roundUp2inN(taskLimit)),
_wantedTaskLimit(_taskLimit.load()),
_rp(0),
@@ -19,10 +23,13 @@ SingleExecutor::SingleExecutor(uint32_t taskLimit)
_wakeupConsumerAt(0),
_producerNeedWakeupAt(0),
_wp(0),
+ _watermark(watermark),
+ _napTime(napTime),
_closed(false)
{
_thread.start();
}
+
SingleExecutor::~SingleExecutor() {
shutdown();
sync();
@@ -102,10 +109,10 @@ SingleExecutor::run() {
while (!_thread.stopped()) {
drain_tasks();
_producerCondition.notify_all();
- _wakeupConsumerAt.store(_wp.load(std::memory_order_relaxed) + (_taskLimit.load(std::memory_order_relaxed) / 4), std::memory_order_relaxed);
+ _wakeupConsumerAt.store(_wp.load(std::memory_order_relaxed) + _watermark, std::memory_order_relaxed);
Lock lock(_mutex);
if (numTasks() <= 0) {
- _consumerCondition.wait_for(lock, 10ms);
+ _consumerCondition.wait_for(lock, _napTime);
}
_wakeupConsumerAt.store(0, std::memory_order_relaxed);
}
@@ -144,7 +151,7 @@ SingleExecutor::wait_for_room(Lock & lock) {
}
_queueSize.add(numTasks());
while (numTasks() >= _taskLimit.load(std::memory_order_relaxed)) {
- sleepProducer(lock, 10ms, wp - taskLimit/4);
+ sleepProducer(lock, _napTime, wp - _watermark);
}
}
diff --git a/staging_vespalib/src/vespa/vespalib/util/singleexecutor.h b/staging_vespalib/src/vespa/vespalib/util/singleexecutor.h
index 3d759769ea3..cb78f8448f4 100644
--- a/staging_vespalib/src/vespa/vespalib/util/singleexecutor.h
+++ b/staging_vespalib/src/vespa/vespalib/util/singleexecutor.h
@@ -19,6 +19,7 @@ namespace vespalib {
class SingleExecutor final : public vespalib::SyncableThreadExecutor, vespalib::Runnable {
public:
explicit SingleExecutor(uint32_t taskLimit);
+ SingleExecutor(uint32_t taskLimit, uint32_t watermark, duration napTime);
~SingleExecutor() override;
Task::UP execute(Task::UP task) override;
void setTaskLimit(uint32_t taskLimit) override;
@@ -56,6 +57,8 @@ private:
std::atomic<uint64_t> _wakeupConsumerAt;
std::atomic<uint64_t> _producerNeedWakeupAt;
std::atomic<uint64_t> _wp;
+ const uint32_t _watermark;
+ const duration _napTime;
bool _closed;
};