From 4199a01adacb3a129bf959d818addf90e140ae91 Mon Sep 17 00:00:00 2001 From: Shashank Pathipati Date: Thu, 23 Jul 2026 10:07:06 +0000 Subject: [PATCH] fix(tme): add multi-pool subscription for ThreadedME to follow thread-split rule table - fixes side-effect of 084f9e8 - fixes MAX_NODE_NUM to be consistent with 64 bit nodemask bitmaps --- doc/reST/cli.rst | 23 ++-- doc/reST/threading.rst | 5 +- source/common/threadpool.cpp | 247 ++++++++++++++++++---------------- source/common/threadpool.h | 4 +- source/encoder/encoder.cpp | 24 ++-- source/encoder/encoder.h | 1 + source/encoder/threadedme.cpp | 70 ++++++++-- source/encoder/threadedme.h | 37 ++++- 8 files changed, 251 insertions(+), 160 deletions(-) diff --git a/doc/reST/cli.rst b/doc/reST/cli.rst index 31304fe90..d391ffbb7 100755 --- a/doc/reST/cli.rst +++ b/doc/reST/cli.rst @@ -306,10 +306,10 @@ Performance Options 64-bit machines, or 16 for 32-bit machines. If the total number of threads in the system doesn't obey this constraint, we may spawn fewer threads than cores which has been empirically shown to be better for performance. - However, when :option:`--threaded-me` is enabled, this behavior is - overridden and a single thread pool larger than 64 threads may be - created. ThreadedME is a singleton job provider, and multiple frame - encoders may push work to it concurrently. + When :option:`--threaded-me` is enabled, small final pools are retained + so its calculated thread allocation is not truncated. ThreadedME remains + a singleton scheduler, but uses proxy job providers when its workers span + multiple physical pools. No physical pool exceeds the bitmap thread limit. If the five pool features: :option:`--wpp`, :option:`--pmode(deprecated)`, :option:`--pme(deprecated)`, :option:`--lookahead-slices` and :option:`--threaded-me` @@ -320,14 +320,13 @@ Performance Options When :option:`--threaded-me` is enabled, x265 estimates the number of threads to assign to ThreadedME based on motion estimation workload and - spawns a dedicated threadpool (threadpool 0) for it. This pool may span - multiple NUMA nodes when the ThreadedME allocation target requires it. The remaining - threads are then used to create the other pools, which are assigned to - frame encoders and lookahead. - - Frame encoders are distributed between the available thread pools, - and the encoder will never generate more thread pools than - :option:`--frame-threads`. The pools are used for WPP and for + assigns one or more leading physical thread pools to it. These pools may + span multiple NUMA nodes when the ThreadedME allocation target requires + it. The remaining pools are assigned to frame encoders and lookahead. + + Frame encoders are distributed between the available frame pools, + and the encoder will never generate more frame pools than + :option:`--frame-threads`. The pools are used for WPP and for distributed analysis and motion search. On Windows, the native APIs offer sufficient functionality to diff --git a/doc/reST/threading.rst b/doc/reST/threading.rst index 20282e55d..a2bf42dc9 100644 --- a/doc/reST/threading.rst +++ b/doc/reST/threading.rst @@ -281,7 +281,10 @@ as external row dependencies are resolved. The steps are: ThreadedME has higher threading demand because it must maintain a lead over WPP to minimize stalls. It computes MVs for CTUs across rows within a frame and across frames across frame encoders. It is therefore recommended primarily for -many-core systems where threading resources can be balanced. +many-core systems where threading resources can be balanced. Its scheduler is +a singleton, but its configured worker allocation may be split across multiple +physical pools through proxy job providers so that each pool remains within its +worker-bitmap limit. SAO === diff --git a/source/common/threadpool.cpp b/source/common/threadpool.cpp index 33ee01753..6d7478ce9 100644 --- a/source/common/threadpool.cpp +++ b/source/common/threadpool.cpp @@ -189,13 +189,13 @@ void WorkerThread::threadMain() SLEEPBITMAP_OR(&m_pool.m_sleepBitmap, idBit); } -void JobProvider::tryWakeOne() +bool JobProvider::tryWakeOne() { int id = m_pool->tryAcquireSleepingThread(SLEEPBITMAP_LOAD(&m_ownerBitmap), ALL_POOL_THREADS); if (id < 0) { m_helpWanted = true; - return; + return false; } WorkerThread& worker = m_pool->m_workers[id]; @@ -207,6 +207,7 @@ void JobProvider::tryWakeOne() SLEEPBITMAP_OR(&worker.m_curJobProvider->m_ownerBitmap, bit); } worker.awaken(); + return true; } int ThreadPool::tryAcquireSleepingThread(sleepbitmap_t firstTryBitmap, sleepbitmap_t secondTryBitmap) @@ -270,16 +271,18 @@ static int getPhysicalPoolCount(int threads) } /* Distributes totalNumThreads between ThreadedME and FrameEncoder pools. - * Modifies threadsPerPool[], nodeMaskPerPool[], numNumaNodes, and numPools in-place. - * Returns the number of threads reserved for frame encoding. */ + * Reorders the resource groups with all ThreadedME groups first and reports + * both logical group and physical pool counts. */ static void distributeThreadsForTme( x265_param* p, int totalNumThreads, - int& numNumaNodes, - bool bNumaSupport, + int& numPoolGroups, + int poolGroupCapacity, int* threadsPerPool, uint64_t* nodeMaskPerPool, int& numPools, + int& numTmePools, + int& numTmePoolGroups, int& threadsFrameEnc) { if (totalNumThreads < MIN_TME_THREADS) @@ -287,12 +290,13 @@ static void distributeThreadsForTme( x265_log(p, X265_LOG_WARNING, "Low thread count detected, disabling --threaded-me." " Minimum recommended is 32 cores / threads\n"); p->bThreadedME = 0; + numTmePools = 0; + numTmePoolGroups = 0; return; } - int targetTME = ThreadPool::configureTmeThreadCount(p, totalNumThreads); - // TME is always assigned to the first pool, and each pool can have at most MAX_POOL_THREADS threads. - targetTME = X265_MIN((targetTME < 1) ? 1 : targetTME, MAX_POOL_THREADS); + int requestedTME = ThreadPool::configureTmeThreadCount(p, totalNumThreads); + int targetTME = (requestedTME < 1) ? 1 : requestedTME; threadsFrameEnc = totalNumThreads - targetTME; int defaultNumFT = ThreadPool::getFrameThreadsCount(p, totalNumThreads); @@ -301,112 +305,74 @@ static void distributeThreadsForTme( threadsFrameEnc = defaultNumFT; targetTME = totalNumThreads - threadsFrameEnc; } + x265_log(p, X265_LOG_INFO, "ThreadedME worker split: %d TME / %d frame worker threads%s\n", + targetTME, threadsFrameEnc, targetTME == requestedTME ? "" : " (adjusted)"); -#if defined(_WIN32_WINNT) && _WIN32_WINNT >= _WIN32_WINNT_WIN7 || HAVE_LIBNUMA - if (bNumaSupport && numNumaNodes > 1) - { - int tmeNumaNodes = 0; - int leftover = 0; - - // First thread pool belongs to ThreadedME - std::vector threads(1, 0); - std::vector nodeMasks(1, 0); - int poolIndex = 0; - - /* Greedily assign whole NUMA nodes to TME until reaching or exceeding the target */ - for (int i = 0; i < numNumaNodes + 1; i++) - { - if (!threadsPerPool[i] && !nodeMaskPerPool[i]) - continue; - - int toTake = X265_MIN(threadsPerPool[i], targetTME - threads[0]); - if (toTake > 0) - { - threads[poolIndex] += toTake; - nodeMasks[poolIndex] |= nodeMaskPerPool[i]; - tmeNumaNodes++; - - if (threads[0] == targetTME) - poolIndex++; + std::vector tmeThreads; + std::vector tmeMasks; + std::vector frameThreads; + std::vector frameMasks; + int remainingTME = targetTME; - if (toTake < threadsPerPool[i]) - leftover = threadsPerPool[i] - toTake; - } - else - { - threads.push_back(threadsPerPool[i]); - nodeMasks.push_back(nodeMaskPerPool[i]); - poolIndex++; - } - } + for (int i = 0; i < numPoolGroups; i++) + { + if (!threadsPerPool[i]) + continue; - // Distribute leftover threads among FrameEncoders - if (leftover) + int toTME = X265_MIN(threadsPerPool[i], remainingTME); + if (toTME) { - // Case 1: There are 1 or more threadpools for FrameEncoder(s) by now - if (threads.size() > 1) - { - int split = static_cast(static_cast(leftover) / (numNumaNodes - 1)); - for (int pool = 1; pool < numNumaNodes; pool++) - { - int give = X265_MIN(split, leftover); - threads[pool] += give; - leftover -= give; - } - } - - // Case 2: FrameEncoder(s) haven't received threads yet - if (threads.size() == 1) - { - threads.push_back(leftover); - // Give the same node mask as the last node of ThreadedME - uint64_t msb = 1; - uint64_t tmeNodeMask = nodeMasks[0]; - while (tmeNodeMask > 1) - { - tmeNodeMask >>= 1; - msb <<= 1; - } - nodeMasks.push_back(msb); - } + tmeThreads.push_back(toTME); + tmeMasks.push_back(nodeMaskPerPool[i]); + remainingTME -= toTME; } - // Apply calculated threadpool assignment - memset(threadsPerPool, 0, sizeof(int) * (numNumaNodes + 2)); - memset(nodeMaskPerPool, 0, sizeof(uint64_t) * (numNumaNodes + 2)); - - numPools = 0; - numNumaNodes = static_cast(threads.size()); - for (int pool = 0; pool < numNumaNodes; pool++) + int toFrame = threadsPerPool[i] - toTME; + if (toFrame) { - threadsPerPool[pool] = threads[pool]; - nodeMaskPerPool[pool] = nodeMasks[pool]; - numPools += getPhysicalPoolCount(threadsPerPool[pool]); + frameThreads.push_back(toFrame); + frameMasks.push_back(nodeMaskPerPool[i]); } } - else -#else - (void) bNumaSupport; -#endif - { - memset(threadsPerPool, 0, sizeof(int) * (numNumaNodes + 2)); - memset(nodeMaskPerPool, 0, sizeof(uint64_t) * (numNumaNodes + 2)); - numPools = 0; + /* If no resource groups were detected, use the auto-detected total. */ + if (remainingTME) + { + tmeThreads.push_back(remainingTME); + tmeMasks.push_back(1); + frameThreads.push_back(threadsFrameEnc); + frameMasks.push_back(1); + } - threadsPerPool[0] = targetTME; - nodeMaskPerPool[0] = 1; - numPools += 1; + memset(threadsPerPool, 0, sizeof(int) * poolGroupCapacity); + memset(nodeMaskPerPool, 0, sizeof(uint64_t) * poolGroupCapacity); - threadsPerPool[1] = threadsFrameEnc; - nodeMaskPerPool[1] = 1; - numPools += getPhysicalPoolCount(threadsFrameEnc); + numPools = 0; + numTmePools = 0; + numTmePoolGroups = static_cast(tmeThreads.size()); + int group = 0; + for (size_t i = 0; i < tmeThreads.size(); i++, group++) + { + X265_CHECK(group < poolGroupCapacity, "ThreadedME pool plan exceeded resource-group capacity\n"); + threadsPerPool[group] = tmeThreads[i]; + nodeMaskPerPool[group] = tmeMasks[i]; + int physicalPools = getPhysicalPoolCount(tmeThreads[i]); + numPools += physicalPools; + numTmePools += physicalPools; + } + for (size_t i = 0; i < frameThreads.size(); i++, group++) + { + X265_CHECK(group < poolGroupCapacity, "ThreadedME pool plan exceeded resource-group capacity\n"); + threadsPerPool[group] = frameThreads[i]; + nodeMaskPerPool[group] = frameMasks[i]; + numPools += getPhysicalPoolCount(frameThreads[i]); } + numPoolGroups = group; } -ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isThreadsReserved) +ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, int& numTmePools, bool isThreadsReserved) { - enum { MAX_NODE_NUM = 127 }; + enum { MAX_NODE_NUM = 64 }; int cpusPerNode[MAX_NODE_NUM + 1]; int threadsPerPool[MAX_NODE_NUM + 2]; uint64_t nodeMaskPerPool[MAX_NODE_NUM + 2]; @@ -416,8 +382,12 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh memset(threadsPerPool, 0, sizeof(threadsPerPool)); memset(nodeMaskPerPool, 0, sizeof(nodeMaskPerPool)); - int numNumaNodes = X265_MIN(getNumaNodeCount(), MAX_NODE_NUM); + int detectedNumaNodes = getNumaNodeCount(); + int numNumaNodes = X265_MIN(detectedNumaNodes, MAX_NODE_NUM); + if (detectedNumaNodes > MAX_NODE_NUM) + x265_log(p, X265_LOG_WARNING, "Only the first %d NUMA nodes can be represented by thread-pool masks\n", MAX_NODE_NUM); bool bNumaSupport = false; + numTmePools = 0; #if defined(_WIN32_WINNT) && _WIN32_WINNT >= _WIN32_WINNT_WIN7 bNumaSupport = true; @@ -461,10 +431,10 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh #if defined(_WIN32_WINNT) && _WIN32_WINNT >= _WIN32_WINNT_WIN7 if (!strlen(p->numaPools) || (strcmp(p->numaPools, "NULL") == 0 || strcmp(p->numaPools, "*") == 0 || strcmp(p->numaPools, "") == 0)) { - char poolString[50] = ""; + char poolString[MAX_NODE_NUM * 12 + 1] = ""; for (int i = 0; i < numNumaNodes; i++) { - char nextCount[10] = ""; + char nextCount[12] = ""; if (i) snprintf(nextCount, sizeof(nextCount), ",%d", cpusPerNode[i]); else @@ -538,17 +508,24 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh for (int i = 0; i < numNumaNodes + 1; i++) totalNumThreads += threadsPerPool[i]; if (!totalNumThreads) + { totalNumThreads = ThreadPool::getCpuCount(); + threadsPerPool[0] = totalNumThreads; + nodeMaskPerPool[0] = 1; + } int threadsFrameEnc = totalNumThreads; - if (p->bThreadedME) + int numPoolGroups = numNumaNodes + 1; + int numTmePoolGroups = 0; + if (p->bThreadedME && !isThreadsReserved) { - distributeThreadsForTme(p, totalNumThreads, numNumaNodes, bNumaSupport, threadsPerPool, - nodeMaskPerPool, numPools, threadsFrameEnc); + distributeThreadsForTme(p, totalNumThreads, numPoolGroups, MAX_NODE_NUM + 2, threadsPerPool, + nodeMaskPerPool, numPools, numTmePools, + numTmePoolGroups, threadsFrameEnc); } // If the last pool size is > MAX_POOL_THREADS, clip it to spawn thread pools only of size >= 1/2 max (heuristic) - if ((threadsPerPool[numNumaNodes] > MAX_POOL_THREADS) && + if (!p->bThreadedME && (threadsPerPool[numNumaNodes] > MAX_POOL_THREADS) && ((threadsPerPool[numNumaNodes] % MAX_POOL_THREADS) < (MAX_POOL_THREADS / 2))) { threadsPerPool[numNumaNodes] -= (threadsPerPool[numNumaNodes] % MAX_POOL_THREADS); @@ -593,25 +570,60 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh numPools = X265_MAX(p->frameNumThreads / 2, 1); } if (isThreadsReserved) + { numPools = 1; + numTmePools = 0; + } + else if (p->bThreadedME) + { + int framePoolCount = numPools - numTmePools; + if (framePoolCount > p->frameNumThreads) + { + int retainedThreads = 0; + int retainedPools = 0; + for (int group = numTmePoolGroups; group < numPoolGroups && retainedPools < p->frameNumThreads; group++) + { + int remaining = threadsPerPool[group]; + while (remaining && retainedPools < p->frameNumThreads) + { + retainedThreads += X265_MIN(remaining, MAX_POOL_THREADS); + remaining -= X265_MIN(remaining, MAX_POOL_THREADS); + retainedPools++; + } + } + x265_log(p, X265_LOG_WARNING, + "ThreadedME frame pools limited by frame threads; %d frame worker threads will not be used\n", + threadsFrameEnc - retainedThreads); + numPools = numTmePools + p->frameNumThreads; + } + } ThreadPool *pools = new ThreadPool[numPools]; if (pools) { - int poolCount = (p->bThreadedME) ? numPools - 1 : numPools; + int poolCount = numPools - numTmePools; int node = 0; for (int i = 0; i < numPools; i++) { - int maxProviders = (p->bThreadedME && i == 0) // threadpool 0 is dedicated to ThreadedME + int maxProviders = (i < numTmePools) ? 1 : (p->frameNumThreads + poolCount - 1) / poolCount + !isThreadsReserved; // +1 is Lookahead, always assigned to threadpool 0 - while (!threadsPerPool[node]) + while (node < numPoolGroups && !threadsPerPool[node]) node++; + if (node == numPoolGroups) + { + x265_log(p, X265_LOG_ERROR, "Thread-pool plan contains fewer resources than physical pools\n"); + delete[] pools; + numPools = 0; + numTmePools = 0; + return NULL; + } // Consume a block no larger than MAX_POOL_THREADS when creating a physical pool. int numThreads = X265_MIN(threadsPerPool[node], MAX_POOL_THREADS); int origNumThreads = numThreads; - if (i == 0 && p->lookaheadThreads > numThreads / 2) + int firstFramePool = numTmePools; + if (i == firstFramePool && p->lookaheadThreads > numThreads / 2) { p->lookaheadThreads = numThreads / 2; x265_log(p, X265_LOG_DEBUG, "Setting lookahead threads to a maximum of half the total number of threads\n"); @@ -622,13 +634,10 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh numThreads = p->lookaheadThreads; maxProviders = 1; } - else if (i == 0) + else if (i == firstFramePool) numThreads -= p->lookaheadThreads; - if (!p->bThreadedME) - { - X265_CHECK(numThreads <= MAX_POOL_THREADS, "a single thread pool cannot have more than MAX_POOL_THREADS threads\n"); - } + X265_CHECK(numThreads <= MAX_POOL_THREADS, "a single thread pool cannot have more than MAX_POOL_THREADS threads\n"); if (!pools[i].create(numThreads, maxProviders, nodeMaskPerPool[node])) { delete[] pools; @@ -642,11 +651,13 @@ ThreadPool* ThreadPool::allocThreadPools(x265_param* p, int& numPools, bool isTh for (int j = 0; j < 64; j++) if ((nodeMaskPerPool[node] >> j) & 1) len += snprintf(nodesstr + len, sizeof(nodesstr) - len, ",%d", j); - x265_log(p, X265_LOG_INFO, "Thread pool %d using %d threads on numa nodes %s\n", i, numThreads, nodesstr + 1); + x265_log(p, X265_LOG_INFO, "%s thread pool %d using %d threads on numa nodes %s\n", + i < numTmePools ? "ThreadedME" : "Frame", i, numThreads, nodesstr + 1); delete[] nodesstr; } else - x265_log(p, X265_LOG_INFO, "Thread pool created using %d threads\n", numThreads); + x265_log(p, X265_LOG_INFO, "%s thread pool %d created using %d threads\n", + i < numTmePools ? "ThreadedME" : "Frame", i, numThreads); threadsPerPool[node] -= origNumThreads; } } @@ -664,7 +675,7 @@ bool ThreadPool::create(int numThreads, int maxProviders, uint64_t nodeMask) { #if defined(_WIN32_WINNT) && _WIN32_WINNT >= _WIN32_WINNT_WIN7 memset(&m_groupAffinity, 0, sizeof(GROUP_AFFINITY)); - for (int i = 0; i < getNumaNodeCount(); i++) + for (int i = 0; i < X265_MIN(getNumaNodeCount(), 64); i++) { int numaNode = ((nodeMask >> i) & 0x1U) ? i : -1; if (numaNode != -1) diff --git a/source/common/threadpool.h b/source/common/threadpool.h index bacd0e1d3..ca402a380 100644 --- a/source/common/threadpool.h +++ b/source/common/threadpool.h @@ -82,7 +82,7 @@ class JobProvider // Will awaken one idle thread, preferring a thread which most recently // performed work for this provider. - void tryWakeOne(); + bool tryWakeOne(); }; class ThreadPool @@ -112,7 +112,7 @@ class ThreadPool 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 ThreadPool* allocThreadPools(x265_param* p, int& numPools, int& numTmePools, bool isThreadsReserved); static int getCpuCount(); static int getNumaNodeCount(); static int getFrameThreadsCount(x265_param* p, int cpuCount); diff --git a/source/encoder/encoder.cpp b/source/encoder/encoder.cpp index 2c8f8b73b..12e38d49f 100644 --- a/source/encoder/encoder.cpp +++ b/source/encoder/encoder.cpp @@ -142,6 +142,7 @@ Encoder::Encoder() m_param = NULL; m_latestParam = NULL; m_threadPool = NULL; + m_numTmePools = 0; m_analysisFileIn = NULL; m_analysisFileOut = NULL; m_filmGrainIn = NULL; @@ -271,7 +272,7 @@ void Encoder::create() m_numPools = 0; if (allowPools) - m_threadPool = ThreadPool::allocThreadPools(p, m_numPools, 0); + m_threadPool = ThreadPool::allocThreadPools(p, m_numPools, m_numTmePools, 0); else { if (!p->frameNumThreads) @@ -334,19 +335,20 @@ void Encoder::create() // First threadpool belongs to ThreadedME, if the feature is enabled if (p->bThreadedME) { - m_threadedME->m_pool = &m_threadPool[0]; - m_threadedME->m_jpId = 0; - - m_threadPool[0].m_numProviders = 1; - m_threadPool[0].m_jpTable[m_threadedME->m_jpId] = m_threadedME; + if (!m_threadedME->bindPools(m_threadPool, m_numTmePools)) + { + x265_log(p, X265_LOG_ERROR, "Failed to bind ThreadedME worker pools\n"); + m_aborted = true; + return; + } } - int numFrameThreadPools = (!m_param->bThreadedME) ? m_numPools : m_numPools - 1; + int numFrameThreadPools = m_numPools - m_numTmePools; for (int i = 0; i < m_param->frameNumThreads; i++) { // Since first pool belongs to ThreadedME - int pool = static_cast(p->bThreadedME) + i % numFrameThreadPools; + int pool = m_numTmePools + i % numFrameThreadPools; m_frameEncoder[i]->m_pool = &m_threadPool[pool]; m_frameEncoder[i]->m_jpId = m_threadPool[pool].m_numProviders++; m_threadPool[pool].m_jpTable[m_frameEncoder[i]->m_jpId] = m_frameEncoder[i]; @@ -379,10 +381,11 @@ void Encoder::create() ThreadPool* lookAheadThreadPool = 0; if (m_param->lookaheadThreads > 0) { - lookAheadThreadPool = ThreadPool::allocThreadPools(p, pools, 1); + int lookaheadTmePools = 0; + lookAheadThreadPool = ThreadPool::allocThreadPools(p, pools, lookaheadTmePools, 1); } else - lookAheadThreadPool = (!m_param->bThreadedME) ? m_threadPool : &m_threadPool[1]; + lookAheadThreadPool = m_threadPool ? &m_threadPool[m_numTmePools] : NULL; m_lookahead = new Lookahead(m_param, lookAheadThreadPool); if (pools) { @@ -6533,4 +6536,3 @@ bool Encoder::computeSPSRPSIndex() } return false; } - diff --git a/source/encoder/encoder.h b/source/encoder/encoder.h index d532f699f..9101b79b1 100644 --- a/source/encoder/encoder.h +++ b/source/encoder/encoder.h @@ -192,6 +192,7 @@ class Encoder : public x265_encoder int m_outputCount; int m_bframeDelay; int m_numPools; + int m_numTmePools; int m_curEncoder; // weighted prediction diff --git a/source/encoder/threadedme.cpp b/source/encoder/threadedme.cpp index 371ef058e..0cdb0aceb 100644 --- a/source/encoder/threadedme.cpp +++ b/source/encoder/threadedme.cpp @@ -32,10 +32,40 @@ namespace X265_NS { int g_puStartIdx[2 * MAX_CU_SIZE + 1][NUM_PART_SIZES] = {{0}}; +TmeProxyProvider::TmeProxyProvider(ThreadedME& master, ThreadPool* pool, int workerOffset) + : m_master(master) + , m_workerOffset(workerOffset) +{ + m_pool = pool; + m_jpId = 0; +} + +void TmeProxyProvider::findJob(int workerThreadId) +{ + m_master.findJob(m_workerOffset + workerThreadId, *this); +} + +bool ThreadedME::bindPools(ThreadPool* pools, int numPools) +{ + int workerOffset = 0; + for (int i = 0; i < numPools; i++) + { + TmeProxyProvider* provider = new TmeProxyProvider(*this, &pools[i], workerOffset); + if (!provider) + return false; + + m_poolProxies.push_back(provider); + pools[i].m_numProviders = 1; + pools[i].m_jpTable[0] = provider; + workerOffset += pools[i].m_numWorkers; + } + m_tldCount = workerOffset; + return !m_poolProxies.empty(); +} + bool ThreadedME::create() { m_active = true; - m_tldCount = m_pool->m_numWorkers; m_tld = new ThreadLocalData[m_tldCount]; for (int i = 0; i < m_tldCount; i++) { @@ -159,26 +189,41 @@ void ThreadedME::threadMain() } } -void ThreadedME::findJob(int workerThreadId) +void ThreadedME::tryWakeOne() +{ + int providerCount = static_cast(m_poolProxies.size()); + for (int i = 0; i < providerCount; i++) + { + int provider = (m_nextProvider + i) % providerCount; + if (m_poolProxies[provider]->tryWakeOne()) + { + m_nextProvider = (provider + 1) % providerCount; + return; + } + } + m_nextProvider = (m_nextProvider + 1) % providerCount; +} + +void ThreadedME::findJob(int workerThreadId, TmeProxyProvider& provider) { m_taskQueueLock.acquire(); if (m_taskQueue.empty()) { + provider.m_helpWanted = false; m_taskQueueLock.release(); - m_helpWanted = true; return; } int64_t stime = x265_mdate(); #ifdef DETAILED_CU_STATS - ScopedElapsedTime tmeTime(m_tld[workerThreadId].analysis.m_stats[m_jpId].tmeTime); - m_tld[workerThreadId].analysis.m_stats[m_jpId].countTmeTasks++; + ScopedElapsedTime tmeTime(m_tld[workerThreadId].analysis.m_stats[0].tmeTime); + m_tld[workerThreadId].analysis.m_stats[0].countTmeTasks++; #endif CTUTask task = m_taskQueue.top(); m_taskQueue.pop(); - m_helpWanted = !m_taskQueue.empty(); + provider.m_helpWanted = !m_taskQueue.empty(); m_taskQueueLock.release(); int numCols = (m_param->sourceWidth + m_param->maxCUSize - 1) / m_param->maxCUSize; @@ -225,16 +270,21 @@ void ThreadedME::stopJobs() void ThreadedME::destroy() { - for (int i = 0; i < m_tldCount; i++) - m_tld[i].destroy(); + if (m_tld) + for (int i = 0; i < m_tldCount; i++) + m_tld[i].destroy(); delete[] m_tld; + m_tld = NULL; + for (size_t i = 0; i < m_poolProxies.size(); i++) + delete m_poolProxies[i]; + m_poolProxies.clear(); } void ThreadedME::collectStats() { #ifdef DETAILED_CU_STATS for (int i = 0; i < m_tldCount; i++) - m_cuStats.accumulate(m_tld[i].analysis.m_stats[m_jpId], *m_param); + m_cuStats.accumulate(m_tld[i].analysis.m_stats[0], *m_param); #endif } @@ -258,4 +308,4 @@ void initCTU(CUData& ctu, int row, int col, CTUTask& task) ctu.initCTU(frame, ctuAddr, slice->m_sliceQp, bFirstRowInSlice, bLastRowInSlice, bLastCuInSlice); } -} \ No newline at end of file +} diff --git a/source/encoder/threadedme.h b/source/encoder/threadedme.h index 07eaf178d..fc69a24a2 100644 --- a/source/encoder/threadedme.h +++ b/source/encoder/threadedme.h @@ -45,6 +45,17 @@ extern int g_puStartIdx[2 * MAX_CU_SIZE + 1][NUM_PART_SIZES]; class Encoder; class Analysis; class FrameEncoder; +class ThreadedME; + +class TmeProxyProvider : public JobProvider +{ +public: + ThreadedME& m_master; + int m_workerOffset; + + TmeProxyProvider(ThreadedME& master, ThreadPool* pool, int workerOffset); + void findJob(int workerThreadId); +}; struct PUBlock { uint32_t width; @@ -154,10 +165,9 @@ struct CompareCTUTask { * @brief Threaded motion-estimation module that schedules CTU blocks across worker threads. * * Owns per-worker analysis state (ThreadLocalData), manages the CTU task queues, - * and exposes a JobProvider interface for the thread pool to execute MVP - * derivation and ME searches in parallel. + * and coordinates proxy job providers across one or more physical pools. */ -class ThreadedME: public JobProvider, public Thread +class ThreadedME: public Thread { public: x265_param* m_param; @@ -172,6 +182,8 @@ class ThreadedME: public JobProvider, public Thread ThreadLocalData* m_tld; int m_tldCount; + std::vector m_poolProxies; + int m_nextProvider; #ifdef DETAILED_CU_STATS CUStats m_cuStats; @@ -180,10 +192,20 @@ class ThreadedME: public JobProvider, public Thread /** * @brief Construct the ThreadedME manager; call create() before use. */ - ThreadedME(x265_param* param, Encoder& enc): m_param(param), m_enc(enc) {}; + ThreadedME(x265_param* param, Encoder& enc) + : m_param(param) + , m_enc(enc) + , m_enqueueSeq(0) + , m_tld(NULL) + , m_tldCount(0) + , m_nextProvider(0) + {}; + + /** Register one proxy provider for every physical ThreadedME pool. */ + bool bindPools(ThreadPool* pools, int numPools); /** - * @brief Creates threadpool, thread local data and registers itself as a job provider + * @brief Creates thread local data for all registered physical pools */ bool create(); @@ -218,7 +240,10 @@ class ThreadedME: public JobProvider, public Thread * * Called by worker threads via JobProvider; processes an entire CTU block. */ - void findJob(int workerThreadId); + void findJob(int workerThreadId, TmeProxyProvider& provider); + + /** Wake a worker from any ThreadedME physical pool. */ + void tryWakeOne(); /** * @brief Stops worker threads