blob: 646dbf0099c07625632ca03bf7f62ba14d808892 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
/**
* This file contains a utility function to handle threads doing a lot of
* single ticks. It brings the following functionality:
*
* - Give application setting up the threads a way to synchronize all the
* threads so it can perform some operation while no thread is ticking.
* - Give multiple threads a way to use common lock for critical region, such
* that you can divide responsible between multiple threads, and still have
* a way to notify and wait for all.
* - Automatically implement registration in deadlock handler, and updating
* tick times there.
* - Give a thread specific context to tick functions, such that one class
* instance can be used for all threads.
* - Hide thread functionality for starting, stopping and running.
* - Minimizes locking by using a single lock that is taken only once per
* tick loop.
*/
#pragma once
#include <memory>
#include <vespa/vespalib/util/time.h>
#include <vespa/vespalib/stllike/string.h>
namespace storage::framework {
struct ThreadPool;
using ThreadIndex = uint32_t;
/**
* \brief Information returned from tick functions to indicate whether thread
* should throttle a bit or not.
*/
class ThreadWaitInfo {
bool _waitWanted;
explicit ThreadWaitInfo(bool waitBeforeNextTick) : _waitWanted(waitBeforeNextTick) {}
public:
static ThreadWaitInfo MORE_WORK_ENQUEUED;
static ThreadWaitInfo NO_MORE_CRITICAL_WORK_KNOWN;
void merge(const ThreadWaitInfo& other);
[[nodiscard]] bool waitWanted() const noexcept { return _waitWanted; }
};
/**
* \brief Simple superclass to implement for ticking threads.
*/
struct TickingThread {
virtual ~TickingThread() = default;
virtual ThreadWaitInfo doCriticalTick(ThreadIndex) = 0;
virtual ThreadWaitInfo doNonCriticalTick(ThreadIndex) = 0;
virtual void newThreadCreated(ThreadIndex) {}
};
/** \brief Delete to allow threads to tick again. */
struct TickingLockGuard {
struct Impl {
virtual ~Impl() = default;
virtual void broadcast() = 0;
};
explicit TickingLockGuard(std::unique_ptr<Impl> impl) : _impl(std::move(impl)) {}
void broadcast() { _impl->broadcast(); }
private:
std::unique_ptr<Impl> _impl;
};
struct ThreadLock {
virtual ~ThreadLock() = default;
virtual TickingLockGuard freezeAllTicks() = 0;
virtual TickingLockGuard freezeCriticalTicks() = 0;
};
/**
* \brief Thread pool set up by the application to control the threads.
*/
struct TickingThreadPool : public ThreadLock {
using UP = std::unique_ptr<TickingThreadPool>;
// TODO STRIPE: Change waitTime default to 100ms when legacy mode is removed.
static TickingThreadPool::UP createDefault(
vespalib::stringref name,
vespalib::duration waitTime,
int ticksBeforeWait,
vespalib::duration maxProcessTime);
static TickingThreadPool::UP createDefault(vespalib::stringref name, vespalib::duration waitTime);
~TickingThreadPool() override = default;
/** All threads must be added before starting the threads. */
virtual void addThread(TickingThread& ticker) = 0;
/** Start all the threads added. */
virtual void start(ThreadPool& pool) = 0;
virtual void stop() = 0;
virtual vespalib::string getStatus() = 0;
};
}
|
