blob: 88e5516d268e998fabcdb9437a56fba9c8fbb955 (
plain)
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
100
101
102
103
104
105
106
107
108
|
// Copyright 2017 Dolphin Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later
#pragma once
#include <functional>
#include <queue>
#include <thread>
#include "Common/Event.h"
#include "Common/Flag.h"
#include "Common/Thread.h"
// A thread that executes the given function for every item placed into its queue.
namespace Common
{
template <typename T>
class WorkQueueThread
{
public:
WorkQueueThread() = default;
WorkQueueThread(std::function<void(T)> function) { Reset(std::move(function)); }
~WorkQueueThread() { Shutdown(); }
void Reset(std::function<void(T)> function)
{
Shutdown();
m_shutdown.Clear();
m_cancelled.Clear();
m_function = std::move(function);
m_thread = std::thread(&WorkQueueThread::ThreadLoop, this);
}
template <typename... Args>
void EmplaceItem(Args&&... args)
{
if (!m_cancelled.IsSet())
{
std::lock_guard lg(m_lock);
m_items.emplace(std::forward<Args>(args)...);
}
m_wakeup.Set();
}
void Clear()
{
{
std::lock_guard lg(m_lock);
m_items = std::queue<T>();
}
m_wakeup.Set();
}
void Cancel()
{
m_cancelled.Set();
Clear();
Shutdown();
}
bool IsCancelled() const { return m_cancelled.IsSet(); }
private:
void Shutdown()
{
if (m_thread.joinable())
{
m_shutdown.Set();
m_wakeup.Set();
m_thread.join();
}
}
void ThreadLoop()
{
Common::SetCurrentThreadName("WorkQueueThread");
while (true)
{
m_wakeup.Wait();
while (true)
{
std::unique_lock lg(m_lock);
if (m_items.empty())
break;
T item{std::move(m_items.front())};
m_items.pop();
lg.unlock();
m_function(std::move(item));
}
if (m_shutdown.IsSet())
break;
}
}
std::function<void(T)> m_function;
std::thread m_thread;
Common::Event m_wakeup;
Common::Flag m_shutdown;
Common::Flag m_cancelled;
std::mutex m_lock;
std::queue<T> m_items;
};
} // namespace Common
|
