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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
/*****************************************************************************
* Copyright (C) 2013-2020 MulticoreWare, Inc
*
* Authors: Steve Borho <steve@borho.org>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02111, USA.
*
* This program is also available under a commercial proprietary license.
* For more information, contact us at license @ x265.com
*****************************************************************************/
#ifndef X265_THREADPOOL_H
#define X265_THREADPOOL_H
#include "common.h"
#include "threading.h"
namespace X265_NS {
// x265 private namespace
class ThreadPool;
class WorkerThread;
class BondedTaskGroup;
#if X86_64
typedef uint64_t sleepbitmap_t;
#else
typedef uint32_t sleepbitmap_t;
#endif
static const sleepbitmap_t ALL_POOL_THREADS = (sleepbitmap_t)-1;
enum { MAX_POOL_THREADS = sizeof(sleepbitmap_t) * 8 };
enum { INVALID_SLICE_PRIORITY = 10 }; // a value larger than any X265_TYPE_* macro
// Frame level job providers. FrameEncoder and Lookahead derive from
// this class and implement findJob()
class JobProvider
{
public:
ThreadPool* m_pool;
sleepbitmap_t m_ownerBitmap;
int m_jpId;
int m_sliceType;
bool m_helpWanted;
bool m_isFrameEncoder; /* rather ugly hack, but nothing better presents itself */
JobProvider()
: m_pool(NULL)
, m_ownerBitmap(0)
, m_jpId(-1)
, m_sliceType(INVALID_SLICE_PRIORITY)
, m_helpWanted(false)
, m_isFrameEncoder(false)
{}
virtual ~JobProvider() {}
// Worker threads will call this method to perform work
virtual void findJob(int workerThreadId) = 0;
// Will awaken one idle thread, preferring a thread which most recently
// performed work for this provider.
void tryWakeOne();
};
class ThreadPool
{
public:
sleepbitmap_t m_sleepBitmap;
int m_numProviders;
int m_numWorkers;
void* m_numaMask; // node mask in linux, cpu mask in windows
#if defined(_WIN32_WINNT) && _WIN32_WINNT >= _WIN32_WINNT_WIN7
GROUP_AFFINITY m_groupAffinity;
#endif
bool m_isActive;
JobProvider** m_jpTable;
WorkerThread* m_workers;
ThreadPool();
~ThreadPool();
bool create(int numThreads, int maxProviders, uint64_t nodeMask);
bool start();
void stopWorkers();
void setCurrentThreadAffinity();
void setThreadNodeAffinity(void *numaMask);
int tryAcquireSleepingThread(sleepbitmap_t firstTryBitmap, sleepbitmap_t secondTryBitmap);
int tryBondPeers(int maxPeers, sleepbitmap_t peerBitmap, BondedTaskGroup& master);
static ThreadPool* allocThreadPools(x265_param* p, int& numPools, bool isThreadsReserved);
static int getCpuCount();
static int getNumaNodeCount();
static void getFrameThreadsCount(x265_param* p,int cpuCount);
};
/* Any worker thread may enlist the help of idle worker threads from the same
* job provider. They must derive from this class and implement the
* processTasks() method. To use, an instance must be instantiated by a worker
* thread (referred to as the master thread) and then tryBondPeers() must be
* called. If it returns non-zero then some number of slave worker threads are
* already in the process of calling your processTasks() function. The master
* thread should participate and call processTasks() itself. When
* waitForExit() returns, all bonded peer threads are guaranteed to have
* exitied processTasks(). Since the thread count is small, it uses explicit
* locking instead of atomic counters and bitmasks */
class BondedTaskGroup
{
public:
Lock m_lock;
ThreadSafeInteger m_exitedPeerCount;
int m_bondedPeerCount;
int m_jobTotal;
int m_jobAcquired;
BondedTaskGroup() { m_bondedPeerCount = m_jobTotal = m_jobAcquired = 0; }
/* Do not allow the instance to be destroyed before all bonded peers have
* exited processTasks() */
~BondedTaskGroup() { waitForExit(); }
/* Try to enlist the help of idle worker threads on most recently associated
* with the given job provider and "bond" them to work on your tasks. Up to
* maxPeers worker threads will call your processTasks() method. */
int tryBondPeers(JobProvider& jp, int maxPeers)
{
int count = jp.m_pool->tryBondPeers(maxPeers, jp.m_ownerBitmap, *this);
m_bondedPeerCount += count;
return count;
}
/* Try to enlist the help of any idle worker threads and "bond" them to work
* on your tasks. Up to maxPeers worker threads will call your
* processTasks() method. */
int tryBondPeers(ThreadPool& pool, int maxPeers)
{
int count = pool.tryBondPeers(maxPeers, ALL_POOL_THREADS, *this);
m_bondedPeerCount += count;
return count;
}
/* Returns when all bonded peers have exited processTasks(). It does *NOT*
* ensure all tasks are completed (but this is generally implied). */
void waitForExit()
{
int exited = m_exitedPeerCount.get();
while (m_bondedPeerCount != exited)
exited = m_exitedPeerCount.waitForChange(exited);
}
/* Derived classes must define this method. The worker thread ID may be
* used to index into thread local data, or ignored. The ID will be between
* 0 and jp.m_numWorkers - 1 */
virtual void processTasks(int workerThreadId) = 0;
};
} // end namespace X265_NS
#endif // ifndef X265_THREADPOOL_H
|
