source/common/threadpool.cpp

   1 /*****************************************************************************
   2  * Copyright (C) 2013 x265 project
   3  *
   4  * Authors: Steve Borho <steve@borho.org>
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 2 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111, USA.
  19  *
  20  * This program is also available under a commercial proprietary license.
  21  * For more information, contact us at license @ x265.com
  22  *****************************************************************************/
  23
  24 #include "common.h"
  25 #include "threadpool.h"
  26 #include "threading.h"
  27
  28 #include <new>
  29
  30 #if MACOS
  31 #include <sys/param.h>
  32 #include <sys/sysctl.h>
  33 #endif
  34
  35 namespace x265 {
  36 // x265 private namespace
  37
  38 class ThreadPoolImpl;
  39
  40 class PoolThread : public Thread
  41 {
  42 private:
  43
  44     ThreadPoolImpl &m_pool;
  45
  46     PoolThread& operator =(const PoolThread&);
  47
  48     int            m_id;
  49
  50     bool           m_dirty;
  51
  52     bool           m_exited;
  53
  54     Event          m_wakeEvent;
  55
  56 public:
  57
  58     PoolThread(ThreadPoolImpl& pool, int id)
  59         : m_pool(pool)
  60         , m_id(id)
  61         , m_dirty(false)
  62         , m_exited(false)
  63     {
  64     }
  65
  66     bool isDirty() const  { return m_dirty; }
  67
  68     void markDirty()      { m_dirty = true; }
  69
  70     bool isExited() const { return m_exited; }
  71
  72     void poke()           { m_wakeEvent.trigger(); }
  73
  74     virtual ~PoolThread() {}
  75
  76     void threadMain();
  77 };
  78
  79 class ThreadPoolImpl : public ThreadPool
  80 {
  81 private:
  82
  83     bool         m_ok;
  84     int          m_referenceCount;
  85     int          m_numThreads;
  86     int          m_numSleepMapWords;
  87     PoolThread  *m_threads;
  88     volatile uint32_t *m_sleepMap;
  89
  90     /* Lock for write access to the provider lists.  Threads are
  91      * always allowed to read m_firstProvider and follow the
  92      * linked list.  Providers must zero their m_nextProvider
  93      * pointers before removing themselves from this list */
  94     Lock         m_writeLock;
  95
  96 public:
  97
  98     static ThreadPoolImpl *s_instance;
  99     static Lock s_createLock;
 100
 101     JobProvider *m_firstProvider;
 102     JobProvider *m_lastProvider;
 103
 104 public:
 105
 106     ThreadPoolImpl(int numthreads);
 107
 108     virtual ~ThreadPoolImpl();
 109
 110     ThreadPoolImpl *AddReference()
 111     {
 112         m_referenceCount++;
 113
 114         return this;
 115     }
 116
 117     void markThreadAsleep(int id);
 118
 119     void waitForAllIdle();
 120
 121     int getThreadCount() const { return m_numThreads; }
 122
 123     bool IsValid() const       { return m_ok; }
 124
 125     void release();
 126
 127     void Stop();
 128
 129     void enqueueJobProvider(JobProvider &);
 130
 131     void dequeueJobProvider(JobProvider &);
 132
 133     void FlushProviderList();
 134
 135     void pokeIdleThread();
 136 };
 137
 138 void PoolThread::threadMain()
 139 {
 140 #if _WIN32
 141     SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_BELOW_NORMAL);
 142 #else
 143     __attribute__((unused)) int val = nice(10);
 144 #endif
 145
 146     while (m_pool.IsValid())
 147     {
 148         /* Walk list of job providers, looking for work */
 149         JobProvider *cur = m_pool.m_firstProvider;
 150         while (cur)
 151         {
 152             // FindJob() may perform actual work and return true.  If
 153             // it does we restart the job search
 154             if (cur->findJob(m_id) == true)
 155                 break;
 156
 157             cur = cur->m_nextProvider;
 158         }
 159
 160         // this thread has reached the end of the provider list
 161         m_dirty = false;
 162
 163         if (cur == NULL)
 164         {
 165             m_pool.markThreadAsleep(m_id);
 166             m_wakeEvent.wait();
 167         }
 168     }
 169
 170     m_exited = true;
 171 }
 172
 173 void ThreadPoolImpl::markThreadAsleep(int id)
 174 {
 175     int word = id >> 5;
 176     uint32_t bit = 1 << (id & 31);
 177
 178     ATOMIC_OR(&m_sleepMap[word], bit);
 179 }
 180
 181 void ThreadPoolImpl::pokeIdleThread()
 182 {
 183     /* Find a bit in the sleeping thread bitmap and poke it awake, do
 184      * not give up until a thread is awakened or all of them are awake */
 185     for (int i = 0; i < m_numSleepMapWords; i++)
 186     {
 187         uint32_t oldval = m_sleepMap[i];
 188         while (oldval)
 189         {
 190             unsigned long id;
 191             CTZ(id, oldval);
 192
 193             uint32_t bit = 1 << id;
 194             if (ATOMIC_AND(&m_sleepMap[i], ~bit) & bit)
 195             {
 196                 m_threads[i * 32 + id].poke();
 197                 return;
 198             }
 199
 200             oldval = m_sleepMap[i];
 201         }
 202     }
 203 }
 204
 205 ThreadPoolImpl *ThreadPoolImpl::s_instance;
 206 Lock ThreadPoolImpl::s_createLock;
 207
 208 /* static */
 209 ThreadPool *ThreadPool::allocThreadPool(int numthreads)
 210 {
 211     if (ThreadPoolImpl::s_instance)
 212         return ThreadPoolImpl::s_instance->AddReference();
 213
 214     /* acquire the lock to create the instance */
 215     ThreadPoolImpl::s_createLock.acquire();
 216
 217     if (ThreadPoolImpl::s_instance)
 218         /* pool was allocated while we waited for the lock */
 219         ThreadPoolImpl::s_instance->AddReference();
 220     else
 221         ThreadPoolImpl::s_instance = new ThreadPoolImpl(numthreads);
 222     ThreadPoolImpl::s_createLock.release();
 223
 224     return ThreadPoolImpl::s_instance;
 225 }
 226
 227 ThreadPool *ThreadPool::getThreadPool()
 228 {
 229     X265_CHECK(ThreadPoolImpl::s_instance, "getThreadPool() called prior to allocThreadPool()\n");
 230     return ThreadPoolImpl::s_instance;
 231 }
 232
 233 void ThreadPoolImpl::release()
 234 {
 235     if (--m_referenceCount == 0)
 236     {
 237         X265_CHECK(this == ThreadPoolImpl::s_instance, "multiple thread pool instances detected\n");
 238         ThreadPoolImpl::s_instance = NULL;
 239         this->Stop();
 240         delete this;
 241     }
 242 }
 243
 244 ThreadPoolImpl::ThreadPoolImpl(int numThreads)
 245     : m_ok(false)
 246     , m_referenceCount(1)
 247     , m_firstProvider(NULL)
 248     , m_lastProvider(NULL)
 249 {
 250     m_numSleepMapWords = (numThreads + 31) >> 5;
 251     m_sleepMap = X265_MALLOC(uint32_t, m_numSleepMapWords);
 252
 253     char *buffer = (char*)X265_MALLOC(PoolThread, numThreads);
 254     m_threads = reinterpret_cast<PoolThread*>(buffer);
 255     m_numThreads = numThreads;
 256
 257     if (m_threads && m_sleepMap)
 258     {
 259         for (int i = 0; i < m_numSleepMapWords; i++)
 260             m_sleepMap[i] = 0;
 261
 262         m_ok = true;
 263         int i;
 264         for (i = 0; i < numThreads; i++)
 265         {
 266             new (buffer)PoolThread(*this, i);
 267             buffer += sizeof(PoolThread);
 268             if (!m_threads[i].start())
 269             {
 270                 m_ok = false;
 271                 break;
 272             }
 273         }
 274
 275         if (m_ok)
 276             waitForAllIdle();
 277         else
 278         {
 279             // stop threads that did start up
 280             for (int j = 0; j < i; j++)
 281             {
 282                 m_threads[j].poke();
 283                 m_threads[j].stop();
 284             }
 285         }
 286     }
 287 }
 288
 289 void ThreadPoolImpl::waitForAllIdle()
 290 {
 291     if (!m_ok)
 292         return;
 293
 294     int id = 0;
 295     do
 296     {
 297         int word = id >> 5;
 298         uint32_t bit = 1 << (id & 31);
 299         if (m_sleepMap[word] & bit)
 300             id++;
 301         else
 302         {
 303             GIVE_UP_TIME();
 304         }
 305     }
 306     while (id < m_numThreads);
 307 }
 308
 309 void ThreadPoolImpl::Stop()
 310 {
 311     if (m_ok)
 312     {
 313         waitForAllIdle();
 314
 315         // set invalid flag, then wake them up so they exit their main func
 316         m_ok = false;
 317         for (int i = 0; i < m_numThreads; i++)
 318         {
 319             m_threads[i].poke();
 320             m_threads[i].stop();
 321         }
 322     }
 323 }
 324
 325 ThreadPoolImpl::~ThreadPoolImpl()
 326 {
 327     X265_FREE((void*)m_sleepMap);
 328
 329     if (m_threads)
 330     {
 331         // cleanup thread handles
 332         for (int i = 0; i < m_numThreads; i++)
 333             m_threads[i].~PoolThread();
 334
 335         X265_FREE(reinterpret_cast<char*>(m_threads));
 336     }
 337 }
 338
 339 void ThreadPoolImpl::enqueueJobProvider(JobProvider &p)
 340 {
 341     // only one list writer at a time
 342     ScopedLock l(m_writeLock);
 343
 344     p.m_nextProvider = NULL;
 345     p.m_prevProvider = m_lastProvider;
 346     m_lastProvider = &p;
 347
 348     if (p.m_prevProvider)
 349         p.m_prevProvider->m_nextProvider = &p;
 350     else
 351         m_firstProvider = &p;
 352 }
 353
 354 void ThreadPoolImpl::dequeueJobProvider(JobProvider &p)
 355 {
 356     // only one list writer at a time
 357     ScopedLock l(m_writeLock);
 358
 359     // update pool entry pointers first
 360     if (m_firstProvider == &p)
 361         m_firstProvider = p.m_nextProvider;
 362
 363     if (m_lastProvider == &p)
 364         m_lastProvider = p.m_prevProvider;
 365
 366     // extract self from doubly linked lists
 367     if (p.m_nextProvider)
 368         p.m_nextProvider->m_prevProvider = p.m_prevProvider;
 369
 370     if (p.m_prevProvider)
 371         p.m_prevProvider->m_nextProvider = p.m_nextProvider;
 372
 373     p.m_nextProvider = NULL;
 374     p.m_prevProvider = NULL;
 375 }
 376
 377 /* Ensure all threads have made a full pass through the provider list, ensuring
 378  * dequeued providers are safe for deletion. */
 379 void ThreadPoolImpl::FlushProviderList()
 380 {
 381     for (int i = 0; i < m_numThreads; i++)
 382     {
 383         m_threads[i].markDirty();
 384         m_threads[i].poke();
 385     }
 386
 387     int i;
 388     do
 389     {
 390         for (i = 0; i < m_numThreads; i++)
 391         {
 392             if (m_threads[i].isDirty())
 393             {
 394                 GIVE_UP_TIME();
 395                 break;
 396             }
 397         }
 398     }
 399     while (i < m_numThreads);
 400 }
 401
 402 void JobProvider::flush()
 403 {
 404     if (m_nextProvider || m_prevProvider)
 405         dequeue();
 406     dynamic_cast<ThreadPoolImpl*>(m_pool)->FlushProviderList();
 407 }
 408
 409 void JobProvider::enqueue()
 410 {
 411     // Add this provider to the end of the thread pool's job provider list
 412     X265_CHECK(!m_nextProvider && !m_prevProvider && m_pool, "job provider was already queued\n");
 413     m_pool->enqueueJobProvider(*this);
 414     m_pool->pokeIdleThread();
 415 }
 416
 417 void JobProvider::dequeue()
 418 {
 419     // Remove this provider from the thread pool's job provider list
 420     m_pool->dequeueJobProvider(*this);
 421     // Ensure no jobs were missed while the provider was being removed
 422     m_pool->pokeIdleThread();
 423 }
 424
 425 int getCpuCount()
 426 {
 427 #if _WIN32
 428     SYSTEM_INFO sysinfo;
 429     GetSystemInfo(&sysinfo);
 430     return sysinfo.dwNumberOfProcessors;
 431 #elif __unix__
 432     return sysconf(_SC_NPROCESSORS_ONLN);
 433 #elif MACOS
 434     int nm[2];
 435     size_t len = 4;
 436     uint32_t count;
 437
 438     nm[0] = CTL_HW;
 439     nm[1] = HW_AVAILCPU;
 440     sysctl(nm, 2, &count, &len, NULL, 0);
 441
 442     if (count < 1)
 443     {
 444         nm[1] = HW_NCPU;
 445         sysctl(nm, 2, &count, &len, NULL, 0);
 446         if (count < 1)
 447             count = 1;
 448     }
 449
 450     return count;
 451 #else // if _WIN32
 452     return 2; // default to 2 threads, everywhere else
 453 #endif // if _WIN32
 454 }
 455 } // end namespace x265