source/common/threading.h

   1 /*****************************************************************************
   2  * x265: threading class and intrinsics
   3  *****************************************************************************
   4  * Copyright (C) 2013 x265 project
   5  *
   6  * Authors: Steve Borho <steve@borho.org>
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 2 of the License, or
  11  * (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111, USA.
  21  *
  22  * This program is also available under a commercial proprietary license.
  23  * For more information, contact us at license @ x265.com
  24  *****************************************************************************/
  25
  26 #ifndef X265_THREADING_H
  27 #define X265_THREADING_H
  28
  29 #include "common.h"
  30 #include "x265.h"
  31
  32 #ifdef _WIN32
  33 #include <windows.h>
  34 #include "winxp.h"  // XP workarounds for CONDITION_VARIABLE and ATOMIC_OR
  35 #else
  36 #include <pthread.h>
  37 #include <semaphore.h>
  38 #include <errno.h>
  39 #include <fcntl.h>
  40 #endif
  41
  42 #if MACOS
  43 #include <sys/param.h>
  44 #include <sys/sysctl.h>
  45 #endif
  46
  47 #ifdef __GNUC__                         /* GCCs builtin atomics */
  48
  49 #include <sys/time.h>
  50 #include <unistd.h>
  51
  52 #define CLZ32(id, x)                        id = (unsigned long)__builtin_clz(x) ^ 31
  53 #define CTZ64(id, x)                        id = (unsigned long)__builtin_ctzll(x)
  54 #define ATOMIC_OR(ptr, mask)                __sync_or_and_fetch(ptr, mask)
  55 #define ATOMIC_CAS(ptr, oldval, newval)     __sync_val_compare_and_swap(ptr, oldval, newval)
  56 #define ATOMIC_CAS32(ptr, oldval, newval)   __sync_val_compare_and_swap(ptr, oldval, newval)
  57 #define ATOMIC_INC(ptr)                     __sync_add_and_fetch((volatile int32_t*)ptr, 1)
  58 #define ATOMIC_DEC(ptr)                     __sync_add_and_fetch((volatile int32_t*)ptr, -1)
  59 #define GIVE_UP_TIME()                      usleep(0)
  60
  61 #elif defined(_MSC_VER)                 /* Windows atomic intrinsics */
  62
  63 #include <intrin.h>
  64
  65 #if !_WIN64
  66 inline int _BitScanReverse64(DWORD *id, uint64_t x64) // fake 64bit CLZ
  67 {
  68     uint32_t high32 = (uint32_t)(x64 >> 32);
  69     uint32_t low32 = (uint32_t)x64;
  70
  71     if (high32)
  72     {
  73         _BitScanReverse(id, high32);
  74         *id += 32;
  75         return 1;
  76     }
  77     else if (low32)
  78         return _BitScanReverse(id, low32);
  79     else
  80         return *id = 0;
  81 }
  82
  83 inline int _BitScanForward64(DWORD *id, uint64_t x64) // fake 64bit CLZ
  84 {
  85     uint32_t high32 = (uint32_t)(x64 >> 32);
  86     uint32_t low32 = (uint32_t)x64;
  87
  88     if (high32)
  89     {
  90         _BitScanForward(id, high32);
  91         *id += 32;
  92         return 1;
  93     }
  94     else if (low32)
  95         return _BitScanForward(id, low32);
  96     else
  97         return *id = 0;
  98 }
  99
 100 #endif // if !_WIN64
 101
 102 #ifndef ATOMIC_OR
 103 #define ATOMIC_OR(ptr, mask)                InterlockedOr64((volatile LONG64*)ptr, mask)
 104 #endif
 105
 106 #define CLZ32(id, x)                        _BitScanReverse(&id, x)
 107 #define CTZ64(id, x)                        _BitScanForward64(&id, x)
 108 #define ATOMIC_CAS(ptr, oldval, newval)     (uint64_t)_InterlockedCompareExchange64((volatile LONG64*)ptr, newval, oldval)
 109 #define ATOMIC_CAS32(ptr, oldval, newval)   (uint64_t)_InterlockedCompareExchange((volatile LONG*)ptr, newval, oldval)
 110 #define ATOMIC_INC(ptr)                     InterlockedIncrement((volatile LONG*)ptr)
 111 #define ATOMIC_DEC(ptr)                     InterlockedDecrement((volatile LONG*)ptr)
 112 #define GIVE_UP_TIME()                      Sleep(0)
 113
 114 #endif // ifdef __GNUC__
 115
 116 namespace x265 {
 117 // x265 private namespace
 118
 119 #ifdef _WIN32
 120
 121 typedef HANDLE ThreadHandle;
 122
 123 class Lock
 124 {
 125 public:
 126
 127     Lock()
 128     {
 129         InitializeCriticalSection(&this->handle);
 130     }
 131
 132     ~Lock()
 133     {
 134         DeleteCriticalSection(&this->handle);
 135     }
 136
 137     void acquire()
 138     {
 139         EnterCriticalSection(&this->handle);
 140     }
 141
 142     void release()
 143     {
 144         LeaveCriticalSection(&this->handle);
 145     }
 146
 147 protected:
 148
 149     CRITICAL_SECTION handle;
 150 };
 151
 152 class Event
 153 {
 154 public:
 155
 156     Event()
 157     {
 158         this->handle = CreateEvent(NULL, FALSE, FALSE, NULL);
 159     }
 160
 161     ~Event()
 162     {
 163         CloseHandle(this->handle);
 164     }
 165
 166     void wait()
 167     {
 168         WaitForSingleObject(this->handle, INFINITE);
 169     }
 170
 171     bool timedWait(uint32_t milliseconds)
 172     {
 173         /* returns true if event was signaled */
 174         return WaitForSingleObject(this->handle, milliseconds) == WAIT_OBJECT_0;
 175     }
 176
 177     void trigger()
 178     {
 179         SetEvent(this->handle);
 180     }
 181
 182 protected:
 183
 184     HANDLE handle;
 185 };
 186
 187 /* This class is intended for use in signaling state changes safely between CPU
 188  * cores. One thread should be a writer and multiple threads may be readers. The
 189  * mutex's main purpose is to serve as a memory fence to ensure writes made by
 190  * the writer thread are visible prior to readers seeing the m_val change. Its
 191  * secondary purpose is for use with the condition variable for blocking waits */
 192 class ThreadSafeInteger
 193 {
 194 public:
 195
 196     ThreadSafeInteger()
 197     {
 198         m_val = 0;
 199         InitializeCriticalSection(&m_cs);
 200         InitializeConditionVariable(&m_cv);
 201     }
 202
 203     ~ThreadSafeInteger()
 204     {
 205         DeleteCriticalSection(&m_cs);
 206         XP_CONDITION_VAR_FREE(&m_cv);
 207     }
 208
 209     int waitForChange(int prev)
 210     {
 211         EnterCriticalSection(&m_cs);
 212         if (m_val == prev)
 213             SleepConditionVariableCS(&m_cv, &m_cs, INFINITE);
 214         LeaveCriticalSection(&m_cs);
 215         return m_val;
 216     }
 217
 218     int get()
 219     {
 220         EnterCriticalSection(&m_cs);
 221         int ret = m_val;
 222         LeaveCriticalSection(&m_cs);
 223         return ret;
 224     }
 225
 226     void set(int newval)
 227     {
 228         EnterCriticalSection(&m_cs);
 229         m_val = newval;
 230         WakeAllConditionVariable(&m_cv);
 231         LeaveCriticalSection(&m_cs);
 232     }
 233
 234     void incr()
 235     {
 236         EnterCriticalSection(&m_cs);
 237         m_val++;
 238         WakeAllConditionVariable(&m_cv);
 239         LeaveCriticalSection(&m_cs);
 240     }
 241
 242 protected:
 243
 244     CRITICAL_SECTION   m_cs;
 245     CONDITION_VARIABLE m_cv;
 246     int                m_val;
 247 };
 248
 249 #else /* POSIX / pthreads */
 250
 251 typedef pthread_t ThreadHandle;
 252
 253 class Lock
 254 {
 255 public:
 256
 257     Lock()
 258     {
 259         pthread_mutex_init(&this->handle, NULL);
 260     }
 261
 262     ~Lock()
 263     {
 264         pthread_mutex_destroy(&this->handle);
 265     }
 266
 267     void acquire()
 268     {
 269         pthread_mutex_lock(&this->handle);
 270     }
 271
 272     void release()
 273     {
 274         pthread_mutex_unlock(&this->handle);
 275     }
 276
 277 protected:
 278
 279     pthread_mutex_t handle;
 280 };
 281
 282 class Event
 283 {
 284 public:
 285
 286     Event()
 287     {
 288         m_counter = 0;
 289         if (pthread_mutex_init(&m_mutex, NULL) ||
 290             pthread_cond_init(&m_cond, NULL))
 291         {
 292             x265_log(NULL, X265_LOG_ERROR, "fatal: unable to initialize conditional variable\n");
 293         }
 294     }
 295
 296     ~Event()
 297     {
 298         pthread_cond_destroy(&m_cond);
 299         pthread_mutex_destroy(&m_mutex);
 300     }
 301
 302     void wait()
 303     {
 304         pthread_mutex_lock(&m_mutex);
 305
 306         /* blocking wait on conditional variable, mutex is atomically released
 307          * while blocked. When condition is signaled, mutex is re-acquired */
 308         while (m_counter == 0)
 309         {
 310             pthread_cond_wait(&m_cond, &m_mutex);
 311         }
 312
 313         m_counter--;
 314         pthread_mutex_unlock(&m_mutex);
 315     }
 316
 317     bool timedWait(uint32_t waitms)
 318     {
 319         bool bTimedOut = false;
 320
 321         pthread_mutex_lock(&m_mutex);
 322         if (m_counter == 0)
 323         {
 324             struct timeval tv;
 325             struct timespec ts;
 326             gettimeofday(&tv, NULL);
 327             /* convert current time from (sec, usec) to (sec, nsec) */
 328             ts.tv_sec = tv.tv_sec;
 329             ts.tv_nsec = tv.tv_usec * 1000;
 330
 331             ts.tv_nsec += 1000 * 1000 * (waitms % 1000);    /* add ms to tv_nsec */
 332             ts.tv_sec += ts.tv_nsec / (1000 * 1000 * 1000); /* overflow tv_nsec */
 333             ts.tv_nsec %= (1000 * 1000 * 1000);             /* clamp tv_nsec */
 334             ts.tv_sec += waitms / 1000;                     /* add seconds */
 335
 336             /* blocking wait on conditional variable, mutex is atomically released
 337              * while blocked. When condition is signaled, mutex is re-acquired.
 338              * ts is absolute time to stop waiting */
 339             bTimedOut = pthread_cond_timedwait(&m_cond, &m_mutex, &ts) == ETIMEDOUT;
 340         }
 341         if (m_counter > 0)
 342             m_counter--;
 343         pthread_mutex_unlock(&m_mutex);
 344         return bTimedOut;
 345     }
 346
 347     void trigger()
 348     {
 349         pthread_mutex_lock(&m_mutex);
 350         if (m_counter < UINT_MAX)
 351             m_counter++;
 352         /* Signal a single blocking thread */
 353         pthread_cond_signal(&m_cond);
 354         pthread_mutex_unlock(&m_mutex);
 355     }
 356
 357 protected:
 358
 359     pthread_mutex_t m_mutex;
 360     pthread_cond_t  m_cond;
 361     uint32_t        m_counter;
 362 };
 363
 364 /* This class is intended for use in signaling state changes safely between CPU
 365  * cores. One thread should be a writer and multiple threads may be readers. The
 366  * mutex's main purpose is to serve as a memory fence to ensure writes made by
 367  * the writer thread are visible prior to readers seeing the m_val change. Its
 368  * secondary purpose is for use with the condition variable for blocking waits */
 369 class ThreadSafeInteger
 370 {
 371 public:
 372
 373     ThreadSafeInteger()
 374     {
 375         m_val = 0;
 376         if (pthread_mutex_init(&m_mutex, NULL) ||
 377             pthread_cond_init(&m_cond, NULL))
 378         {
 379             x265_log(NULL, X265_LOG_ERROR, "fatal: unable to initialize conditional variable\n");
 380         }
 381     }
 382
 383     ~ThreadSafeInteger()
 384     {
 385         pthread_cond_destroy(&m_cond);
 386         pthread_mutex_destroy(&m_mutex);
 387     }
 388
 389     int waitForChange(int prev)
 390     {
 391         pthread_mutex_lock(&m_mutex);
 392         if (m_val == prev)
 393             pthread_cond_wait(&m_cond, &m_mutex);
 394         pthread_mutex_unlock(&m_mutex);
 395         return m_val;
 396     }
 397
 398     int get()
 399     {
 400         pthread_mutex_lock(&m_mutex);
 401         int ret = m_val;
 402         pthread_mutex_unlock(&m_mutex);
 403         return ret;
 404     }
 405
 406     void set(int newval)
 407     {
 408         pthread_mutex_lock(&m_mutex);
 409         m_val = newval;
 410         pthread_cond_broadcast(&m_cond);
 411         pthread_mutex_unlock(&m_mutex);
 412     }
 413
 414     void incr()
 415     {
 416         pthread_mutex_lock(&m_mutex);
 417         m_val++;
 418         pthread_cond_broadcast(&m_cond);
 419         pthread_mutex_unlock(&m_mutex);
 420     }
 421
 422 protected:
 423
 424     pthread_mutex_t m_mutex;
 425     pthread_cond_t  m_cond;
 426     int             m_val;
 427 };
 428
 429 #endif // ifdef _WIN32
 430
 431 class ScopedLock
 432 {
 433 public:
 434
 435     ScopedLock(Lock &instance) : inst(instance)
 436     {
 437         this->inst.acquire();
 438     }
 439
 440     ~ScopedLock()
 441     {
 442         this->inst.release();
 443     }
 444
 445 protected:
 446
 447     // do not allow assignments
 448     ScopedLock &operator =(const ScopedLock &);
 449
 450     Lock &inst;
 451 };
 452
 453 //< Simplistic portable thread class.  Shutdown signalling left to derived class
 454 class Thread
 455 {
 456 private:
 457
 458     ThreadHandle thread;
 459
 460 public:
 461
 462     Thread();
 463
 464     virtual ~Thread();
 465
 466     //< Derived class must implement ThreadMain.
 467     virtual void threadMain() = 0;
 468
 469     //< Returns true if thread was successfully created
 470     bool start();
 471
 472     void stop();
 473 };
 474 } // end namespace x265
 475
 476 #endif // ifndef X265_THREADING_H