interpretor/preprocessor/faust-0.9.47mr3/architecture/scheduler.h

   1
   2 #include <stdlib.h>
   3 #include <assert.h>
   4 #include <pthread.h>
   5 #include <stdio.h>
   6 #include <errno.h>
   7 #include <string.h>
   8 #include <semaphore.h>
   9 #include <sys/types.h>
  10 #include <sys/stat.h>
  11 #include <errno.h>
  12 #include <fcntl.h>
  13 #include <unistd.h>
  14 #include <math.h>
  15
  16 using namespace std;
  17
  18 // Globals
  19
  20 #define THREAD_SIZE 64
  21 #define QUEUE_SIZE 4096
  22
  23 #define WORK_STEALING_INDEX 0
  24 #define LAST_TASK_INDEX 1
  25
  26
  27 #ifdef __ICC
  28 #define INLINE __forceinline
  29 #else
  30 #define INLINE inline
  31 #endif
  32
  33
  34 // On Intel set FZ (Flush to Zero) and DAZ (Denormals Are Zero)
  35 // flags to avoid costly denormals
  36 #ifdef __SSE__
  37 #include <xmmintrin.h>
  38 #ifdef __SSE2__
  39 #define AVOIDDENORMALS _mm_setcsr(_mm_getcsr() | 0x8040)
  40 #else
  41 #define AVOIDDENORMALS _mm_setcsr(_mm_getcsr() | 0x8000)
  42 #endif
  43 #else
  44 #define AVOIDDENORMALS
  45 #endif
  46
  47 #ifdef __linux__
  48
  49 // handle 32/64 bits int size issues
  50 #ifdef __x86_64__
  51 #define UInt32  unsigned int
  52 #define UInt64  unsigned long int
  53 #else
  54 #define UInt32  unsigned int
  55 #define UInt64  unsigned long long int
  56 #endif
  57
  58 #endif
  59
  60 #ifdef __APPLE__
  61 #include <CoreServices/../Frameworks/CarbonCore.framework/Headers/MacTypes.h>
  62 #endif
  63
  64 class TaskQueue;
  65 struct DSPThreadPool;
  66
  67 extern TaskQueue* gTaskQueueList[THREAD_SIZE];
  68 extern DSPThreadPool* gThreadPool;
  69 extern int gClientCount;
  70 extern UInt64 gMaxStealing;
  71
  72 void Yield();
  73
  74 /**
  75  * Returns the number of clock cycles elapsed since the last reset
  76  * of the processor
  77  */
  78 static INLINE UInt64 DSP_rdtsc(void)
  79 {
  80         union {
  81                 UInt32 i32[2];
  82                 UInt64 i64;
  83         } count;
  84
  85         __asm__ __volatile__("rdtsc" : "=a" (count.i32[0]), "=d" (count.i32[1]));
  86      return count.i64;
  87 }
  88
  89 #if defined(__i386__) || defined(__x86_64__)
  90
  91 #define LOCK "lock ; "
  92
  93 static INLINE void NOP(void)
  94 {
  95         __asm__ __volatile__("nop \n\t");
  96 }
  97
  98 static INLINE char CAS1(volatile void* addr, volatile int value, int newvalue)
  99 {
 100     register char ret;
 101     __asm__ __volatile__ (
 102                                                   "# CAS \n\t"
 103                                                   LOCK "cmpxchg %2, (%1) \n\t"
 104                                                   "sete %0               \n\t"
 105                                                   : "=a" (ret)
 106                                                   : "c" (addr), "d" (newvalue), "a" (value)
 107                           : "memory"
 108                                                   );
 109     return ret;
 110 }
 111
 112 static INLINE int atomic_xadd(volatile int* atomic, int val)
 113 {
 114     register int result;
 115     __asm__ __volatile__ ("# atomic_xadd \n\t"
 116                           LOCK "xaddl %0,%1 \n\t"
 117                           : "=r" (result), "=m" (*atomic)
 118                           : "0" (val), "m" (*atomic));
 119     return result;
 120 }
 121
 122 #endif
 123
 124
 125 /*
 126 static INLINE int INC_ATOMIC(volatile int* val)
 127 {
 128     int actual;
 129     do {
 130         actual = *val;
 131     } while (!CAS1(val, actual, actual + 1));
 132     return actual;
 133 }
 134
 135 static INLINE int DEC_ATOMIC(volatile int* val)
 136 {
 137     int actual;
 138     do {
 139         actual = *val;
 140     } while (!CAS1(val, actual, actual - 1));
 141     return actual;
 142 }
 143 */
 144
 145 static INLINE int INC_ATOMIC(volatile int* val)
 146 {
 147     return atomic_xadd(val, 1);
 148 }
 149
 150 static INLINE int DEC_ATOMIC(volatile int* val)
 151 {
 152     return atomic_xadd(val, -1);
 153 }
 154
 155 // To be used in lock-free queue
 156 struct AtomicCounter
 157 {
 158     union {
 159         struct {
 160             short fHead;
 161             short fTail;
 162         }
 163         scounter;
 164         int fValue;
 165     }info;
 166
 167         INLINE AtomicCounter()
 168         {
 169         info.fValue = 0;
 170     }
 171
 172         INLINE  AtomicCounter& operator=(AtomicCounter& obj)
 173     {
 174         info.fValue = obj.info.fValue;
 175         return *this;
 176     }
 177
 178         INLINE  AtomicCounter& operator=(volatile AtomicCounter& obj)
 179         {
 180         info.fValue = obj.info.fValue;
 181         return *this;
 182     }
 183
 184 };
 185
 186 int get_max_cpu()
 187 {
 188     return sysconf(_SC_NPROCESSORS_ONLN);
 189 }
 190
 191 static int GetPID()
 192 {
 193 #ifdef WIN32
 194     return  _getpid();
 195 #else
 196     return getpid();
 197 #endif
 198 }
 199
 200 #define Value(e) (e).info.fValue
 201
 202 #define Head(e) (e).info.scounter.fHead
 203 #define IncHead(e) (e).info.scounter.fHead++
 204 #define DecHead(e) (e).info.scounter.fHead--
 205
 206 #define Tail(e) (e).info.scounter.fTail
 207 #define IncTail(e) (e).info.scounter.fTail++
 208 #define DecTail(e) (e).info.scounter.fTail--
 209
 210 #define MASTER_THREAD 0
 211
 212 #define MAX_STEAL_DUR 50                    // in usec
 213 #define DEFAULT_CLOCKSPERSEC 2500000000     // in cycles (2,5 Ghz)
 214
 215 class TaskQueue
 216 {
 217     private:
 218
 219         int fTaskList[QUEUE_SIZE];
 220         volatile AtomicCounter fCounter;
 221         UInt64 fStealingStart;
 222
 223     public:
 224
 225         INLINE TaskQueue(int cur_thread)
 226         {
 227             for (int i = 0; i < QUEUE_SIZE; i++) {
 228                 fTaskList[i] = -1;
 229             }
 230             gTaskQueueList[cur_thread] = this;
 231             fStealingStart = 0;
 232         }
 233
 234         INLINE void PushHead(int item)
 235         {
 236             fTaskList[Head(fCounter)] = item;
 237             IncHead(fCounter);
 238         }
 239
 240         INLINE int PopHead()
 241         {
 242             AtomicCounter old_val;
 243             AtomicCounter new_val;
 244
 245             do {
 246                 old_val = fCounter;
 247                 new_val = old_val;
 248                 if (Head(old_val) == Tail(old_val)) {
 249                     return WORK_STEALING_INDEX;
 250                 } else {
 251                     DecHead(new_val);
 252                 }
 253             } while (!CAS1(&fCounter, Value(old_val), Value(new_val)));
 254
 255             return fTaskList[Head(old_val) - 1];
 256         }
 257
 258         INLINE int PopTail()
 259         {
 260             AtomicCounter old_val;
 261             AtomicCounter new_val;
 262
 263             do {
 264                 old_val = fCounter;
 265                 new_val = old_val;
 266                 if (Head(old_val) == Tail(old_val)) {
 267                    return WORK_STEALING_INDEX;
 268                 } else {
 269                     IncTail(new_val);
 270                 }
 271             } while (!CAS1(&fCounter, Value(old_val), Value(new_val)));
 272
 273             return fTaskList[Tail(old_val)];
 274         }
 275
 276                 INLINE void MeasureStealingDur()
 277                 {
 278             // Takes first timetamp
 279             if (fStealingStart == 0) {
 280                 fStealingStart = DSP_rdtsc();
 281             } else if ((DSP_rdtsc() - fStealingStart) > gMaxStealing) {
 282                 Yield();
 283             }
 284                 }
 285
 286                 INLINE void ResetStealingDur()
 287                 {
 288             fStealingStart = 0;
 289                 }
 290
 291         static INLINE int GetNextTask(int thread, int num_threads)
 292         {
 293             int tasknum;
 294             for (int i = 0; i < num_threads; i++) {
 295                 if ((i != thread) && gTaskQueueList[i] && (tasknum = gTaskQueueList[i]->PopTail()) != WORK_STEALING_INDEX) {
 296                 #ifdef __linux__
 297                                         //if (thread != MASTER_THREAD)
 298                                                 gTaskQueueList[thread]->ResetStealingDur();
 299                 #endif
 300                     return tasknum;    // Task is found
 301                 }
 302             }
 303             NOP();
 304           #ifdef __linux__
 305                         //if (thread != MASTER_THREAD)
 306                                 gTaskQueueList[thread]->MeasureStealingDur();
 307         #endif
 308             return WORK_STEALING_INDEX;    // Otherwise will try "workstealing" again next cycle...
 309         }
 310
 311         INLINE void InitTaskList(int task_list_size, int* task_list, int thread_num, int cur_thread, int& tasknum)
 312         {
 313             int task_slice = task_list_size / thread_num;
 314             int task_slice_rest = task_list_size % thread_num;
 315
 316             if (task_slice == 0) {
 317                 // Each thread directly executes one task
 318                 tasknum = task_list[cur_thread];
 319                 // Thread 0 takes remaining ready tasks
 320                 if (cur_thread == 0) {
 321                     for (int index = 0; index < task_slice_rest - thread_num; index++) {
 322                         PushHead(task_list[task_slice_rest + index]);
 323                     }
 324                 }
 325             } else {
 326                 // Each thread takes a part of ready tasks
 327                 int index;
 328                 for (index = 0; index < task_slice - 1; index++) {
 329                     PushHead(task_list[cur_thread * task_slice + index]);
 330                 }
 331                 // Each thread directly executes one task
 332                 tasknum = task_list[cur_thread * task_slice + index];
 333                 // Thread 0 takes remaining ready tasks
 334                 if (cur_thread == 0) {
 335                     for (index = 0; index < task_slice_rest; index++) {
 336                         PushHead(task_list[thread_num * task_slice + index]);
 337                     }
 338                 }
 339             }
 340         }
 341
 342         static INLINE void Init()
 343         {
 344             for (int i = 0; i < THREAD_SIZE; i++) {
 345                 gTaskQueueList[i] = 0;
 346             }
 347         }
 348
 349 };
 350
 351 struct TaskGraph
 352 {
 353     volatile int gTaskList[QUEUE_SIZE];
 354
 355     TaskGraph()
 356     {
 357         for (int i = 0; i < QUEUE_SIZE; i++) {
 358             gTaskList[i] = 0;
 359         }
 360     }
 361
 362     INLINE void InitTask(int task, int val)
 363     {
 364         gTaskList[task] = val;
 365     }
 366
 367     void Display()
 368     {
 369         for (int i = 0; i < QUEUE_SIZE; i++) {
 370             printf("Task = %d activation = %d\n", i, gTaskList[i]);
 371         }
 372     }
 373
 374     INLINE void ActivateOutputTask(TaskQueue& queue, int task, int& tasknum)
 375     {
 376         if (DEC_ATOMIC(&gTaskList[task]) == 1) {
 377             if (tasknum == WORK_STEALING_INDEX) {
 378                 tasknum = task;
 379             } else {
 380                 queue.PushHead(task);
 381             }
 382         }
 383     }
 384
 385     INLINE void ActivateOutputTask(TaskQueue& queue, int task)
 386     {
 387         if (DEC_ATOMIC(&gTaskList[task]) == 1) {
 388             queue.PushHead(task);
 389         }
 390     }
 391
 392     INLINE void ActivateOneOutputTask(TaskQueue& queue, int task, int& tasknum)
 393     {
 394         if (DEC_ATOMIC(&gTaskList[task]) == 1) {
 395             tasknum = task;
 396         } else {
 397             tasknum = queue.PopHead();
 398         }
 399     }
 400
 401     INLINE void GetReadyTask(TaskQueue& queue, int& tasknum)
 402     {
 403         if (tasknum == WORK_STEALING_INDEX) {
 404             tasknum = queue.PopHead();
 405         }
 406     }
 407
 408 };
 409
 410
 411 #define THREAD_POOL_SIZE 16
 412 #define JACK_SCHED_POLICY SCHED_FIFO
 413
 414 /* use 512KB stack per thread - the default is way too high to be feasible
 415  * with mlockall() on many systems */
 416 #define THREAD_STACK 524288
 417
 418
 419 #ifdef __APPLE__
 420
 421 #include <CoreServices/../Frameworks/CarbonCore.framework/Headers/MacTypes.h>
 422 #include <mach/thread_policy.h>
 423 #include <mach/thread_act.h>
 424
 425 #define THREAD_SET_PRIORITY         0
 426 #define THREAD_SCHEDULED_PRIORITY   1
 427
 428 static UInt32 GetThreadPriority(pthread_t thread, int inWhichPriority);
 429
 430 // returns the thread's priority as it was last set by the API
 431 static UInt32 GetThreadSetPriority(pthread_t thread)
 432 {
 433     return GetThreadPriority(thread, THREAD_SET_PRIORITY);
 434 }
 435
 436 // returns the thread's priority as it was last scheduled by the Kernel
 437 static UInt32 GetThreadScheduledPriority(pthread_t thread)
 438 {
 439     return GetThreadPriority(thread, THREAD_SCHEDULED_PRIORITY);
 440 }
 441
 442 static int SetThreadToPriority(pthread_t thread, UInt32 inPriority, Boolean inIsFixed, UInt64 period, UInt64 computation, UInt64 constraint)
 443 {
 444     if (inPriority == 96) {
 445         // REAL-TIME / TIME-CONSTRAINT THREAD
 446         thread_time_constraint_policy_data_t theTCPolicy;
 447         theTCPolicy.period = period;
 448         theTCPolicy.computation = computation;
 449         theTCPolicy.constraint = constraint;
 450         theTCPolicy.preemptible = true;
 451         kern_return_t res = thread_policy_set(pthread_mach_thread_np(thread), THREAD_TIME_CONSTRAINT_POLICY, (thread_policy_t)&theTCPolicy, THREAD_TIME_CONSTRAINT_POLICY_COUNT);
 452         return (res == KERN_SUCCESS) ? 0 : -1;
 453     } else {
 454         // OTHER THREADS
 455         thread_extended_policy_data_t theFixedPolicy;
 456         thread_precedence_policy_data_t thePrecedencePolicy;
 457         SInt32 relativePriority;
 458
 459         // [1] SET FIXED / NOT FIXED
 460         theFixedPolicy.timeshare = !inIsFixed;
 461         thread_policy_set(pthread_mach_thread_np(thread), THREAD_EXTENDED_POLICY, (thread_policy_t)&theFixedPolicy, THREAD_EXTENDED_POLICY_COUNT);
 462
 463         // [2] SET PRECEDENCE
 464         // N.B.: We expect that if thread A created thread B, and the program wishes to change
 465         // the priority of thread B, then the call to change the priority of thread B must be
 466         // made by thread A.
 467         // This assumption allows us to use pthread_self() to correctly calculate the priority
 468         // of the feeder thread (since precedency policy's importance is relative to the
 469         // spawning thread's priority.)
 470         relativePriority = inPriority - GetThreadSetPriority(pthread_self());
 471
 472         thePrecedencePolicy.importance = relativePriority;
 473         kern_return_t res = thread_policy_set(pthread_mach_thread_np(thread), THREAD_PRECEDENCE_POLICY, (thread_policy_t)&thePrecedencePolicy, THREAD_PRECEDENCE_POLICY_COUNT);
 474         return (res == KERN_SUCCESS) ? 0 : -1;
 475     }
 476 }
 477
 478 static UInt32 GetThreadPriority(pthread_t thread, int inWhichPriority)
 479 {
 480     thread_basic_info_data_t threadInfo;
 481     policy_info_data_t thePolicyInfo;
 482     unsigned int count;
 483
 484     // get basic info
 485     count = THREAD_BASIC_INFO_COUNT;
 486     thread_info(pthread_mach_thread_np(thread), THREAD_BASIC_INFO, (thread_info_t)&threadInfo, &count);
 487
 488     switch (threadInfo.policy) {
 489         case POLICY_TIMESHARE:
 490             count = POLICY_TIMESHARE_INFO_COUNT;
 491             thread_info(pthread_mach_thread_np(thread), THREAD_SCHED_TIMESHARE_INFO, (thread_info_t)&(thePolicyInfo.ts), &count);
 492             if (inWhichPriority == THREAD_SCHEDULED_PRIORITY) {
 493                 return thePolicyInfo.ts.cur_priority;
 494             } else {
 495                 return thePolicyInfo.ts.base_priority;
 496             }
 497             break;
 498
 499         case POLICY_FIFO:
 500             count = POLICY_FIFO_INFO_COUNT;
 501             thread_info(pthread_mach_thread_np(thread), THREAD_SCHED_FIFO_INFO, (thread_info_t)&(thePolicyInfo.fifo), &count);
 502             if ((thePolicyInfo.fifo.depressed) && (inWhichPriority == THREAD_SCHEDULED_PRIORITY)) {
 503                 return thePolicyInfo.fifo.depress_priority;
 504             }
 505             return thePolicyInfo.fifo.base_priority;
 506             break;
 507
 508         case POLICY_RR:
 509             count = POLICY_RR_INFO_COUNT;
 510             thread_info(pthread_mach_thread_np(thread), THREAD_SCHED_RR_INFO, (thread_info_t)&(thePolicyInfo.rr), &count);
 511             if ((thePolicyInfo.rr.depressed) && (inWhichPriority == THREAD_SCHEDULED_PRIORITY)) {
 512                 return thePolicyInfo.rr.depress_priority;
 513             }
 514             return thePolicyInfo.rr.base_priority;
 515             break;
 516     }
 517
 518     return 0;
 519 }
 520
 521 static int GetParams(pthread_t thread, UInt64* period, UInt64* computation, UInt64* constraint)
 522 {
 523     thread_time_constraint_policy_data_t theTCPolicy;
 524     mach_msg_type_number_t count = THREAD_TIME_CONSTRAINT_POLICY_COUNT;
 525     boolean_t get_default = false;
 526
 527     kern_return_t res = thread_policy_get(pthread_mach_thread_np(thread),
 528                                           THREAD_TIME_CONSTRAINT_POLICY,
 529                                           (thread_policy_t)&theTCPolicy,
 530                                           &count,
 531                                           &get_default);
 532     if (res == KERN_SUCCESS) {
 533         *period = theTCPolicy.period;
 534         *computation = theTCPolicy.computation;
 535         *constraint = theTCPolicy.constraint;
 536         return 0;
 537     } else {
 538         return -1;
 539     }
 540 }
 541
 542 static UInt64 period = 0;
 543 static UInt64 computation = 0;
 544 static UInt64 constraint = 0;
 545
 546 INLINE void GetRealTime()
 547 {
 548     if (period == 0) {
 549         GetParams(pthread_self(), &period, &computation, &constraint);
 550     }
 551 }
 552
 553 INLINE void SetRealTime()
 554 {
 555     SetThreadToPriority(pthread_self(), 96, true, period, computation, constraint);
 556 }
 557
 558 void CancelThread(pthread_t fThread)
 559 {
 560     mach_port_t machThread = pthread_mach_thread_np(fThread);
 561     thread_terminate(machThread);
 562 }
 563
 564 INLINE void Yield()
 565 {
 566     //sched_yield();
 567 }
 568
 569 #endif
 570
 571 #ifdef __linux__
 572
 573 static int faust_sched_policy = -1;
 574 static struct sched_param faust_rt_param;
 575
 576 INLINE void GetRealTime()
 577 {
 578     if (faust_sched_policy == -1) {
 579         memset(&faust_rt_param, 0, sizeof(faust_rt_param));
 580         pthread_getschedparam(pthread_self(), &faust_sched_policy, &faust_rt_param);
 581     }
 582 }
 583
 584 INLINE void SetRealTime()
 585 {
 586         faust_rt_param.sched_priority--;
 587     pthread_setschedparam(pthread_self(), faust_sched_policy, &faust_rt_param);
 588 }
 589
 590 void CancelThread(pthread_t fThread)
 591 {
 592     pthread_cancel(fThread);
 593     pthread_join(fThread, NULL);
 594 }
 595
 596 INLINE void Yield()
 597 {
 598     pthread_yield();
 599 }
 600
 601
 602 #endif
 603
 604 #define KDSPMESURE 50
 605
 606 static INLINE int Range(int min, int max, int val)
 607 {
 608     if (val < min) {
 609         return min;
 610     } else if (val > max) {
 611         return max;
 612     } else {
 613         return val;
 614     }
 615 }
 616
 617 struct Runnable {
 618
 619     UInt64 fTiming[KDSPMESURE];
 620     UInt64 fStart;
 621     UInt64 fStop;
 622     int fCounter;
 623     float fOldMean;
 624     int fOldfDynamicNumThreads;
 625     bool fDynAdapt;
 626
 627     virtual void computeThread(int cur_thread) = 0;
 628
 629     Runnable():fCounter(0), fOldMean(1000000000.f), fOldfDynamicNumThreads(1)
 630     {
 631         memset(fTiming, 0, sizeof(long long int ) * KDSPMESURE);
 632         fDynAdapt = getenv("OMP_DYN_THREAD") ? strtol(getenv("OMP_DYN_THREAD"), NULL, 10) : false;
 633     }
 634
 635     INLINE float ComputeMean()
 636     {
 637         float mean = 0;
 638         for (int i = 0; i < KDSPMESURE; i++) {
 639             mean += float(fTiming[i]);
 640         }
 641         mean /= float(KDSPMESURE);
 642         return mean;
 643     }
 644
 645     INLINE void StartMeasure()
 646     {
 647         if (!fDynAdapt)
 648             return;
 649
 650         fStart = DSP_rdtsc();
 651     }
 652
 653     INLINE void StopMeasure(int staticthreadnum, int& dynthreadnum)
 654     {
 655         if (!fDynAdapt)
 656             return;
 657
 658         fStop = DSP_rdtsc();
 659         fCounter = (fCounter + 1) % KDSPMESURE;
 660         if (fCounter == 0) {
 661             float mean = ComputeMean();
 662             if (fabs(mean - fOldMean) > 5000) {
 663                 if (mean > fOldMean) { // Worse...
 664                     //printf("Worse %f %f\n", mean, fOldMean);
 665                     if (fOldfDynamicNumThreads > dynthreadnum) {
 666                         fOldfDynamicNumThreads = dynthreadnum;
 667                         dynthreadnum += 1;
 668                     } else {
 669                         fOldfDynamicNumThreads = dynthreadnum;
 670                         dynthreadnum -= 1;
 671                     }
 672                  } else { // Better...
 673                     //printf("Better %f %f\n", mean, fOldMean);
 674                     if (fOldfDynamicNumThreads > dynthreadnum) {
 675                         fOldfDynamicNumThreads = dynthreadnum;
 676                         dynthreadnum -= 1;
 677                     } else {
 678                         fOldfDynamicNumThreads = dynthreadnum;
 679                         dynthreadnum += 1;
 680                     }
 681                 }
 682                 fOldMean = mean;
 683                 dynthreadnum = Range(1, staticthreadnum, dynthreadnum);
 684                 //printf("dynthreadnum %d\n", dynthreadnum);
 685             }
 686         }
 687         fTiming[fCounter] = fStop - fStart;
 688     }
 689 };
 690
 691 struct DSPThread;
 692
 693 struct DSPThreadPool {
 694
 695     DSPThread* fThreadPool[THREAD_POOL_SIZE];
 696     int fThreadCount;
 697     volatile int fCurThreadCount;
 698
 699     DSPThreadPool();
 700     ~DSPThreadPool();
 701
 702     void StartAll(int num, bool realtime);
 703     void StopAll();
 704     void SignalAll(int num, Runnable* runnable);
 705
 706     void SignalOne();
 707     bool IsFinished();
 708
 709     static DSPThreadPool* Init();
 710     static void Destroy();
 711
 712 };
 713
 714 struct DSPThread {
 715
 716     pthread_t fThread;
 717     DSPThreadPool* fThreadPool;
 718     Runnable* fRunnable;
 719     sem_t* fSemaphore;
 720     char fName[128];
 721     bool fRealTime;
 722     int fNum;
 723
 724     DSPThread(int num, DSPThreadPool* pool)
 725     {
 726         fNum = num;
 727         fThreadPool = pool;
 728         fRunnable = NULL;
 729         fRealTime = false;
 730
 731         sprintf(fName, "faust_sem_%d_%p", GetPID(), this);
 732
 733         if ((fSemaphore = sem_open(fName, O_CREAT, 0777, 0)) == (sem_t*)SEM_FAILED) {
 734             printf("Allocate: can't check in named semaphore name = %s err = %s", fName, strerror(errno));
 735         }
 736     }
 737
 738     virtual ~DSPThread()
 739     {
 740         sem_unlink(fName);
 741         sem_close(fSemaphore);
 742     }
 743
 744     void Run()
 745     {
 746         while (sem_wait(fSemaphore) != 0) {}
 747         fRunnable->computeThread(fNum + 1);
 748         fThreadPool->SignalOne();
 749     }
 750
 751     static void* ThreadHandler(void* arg)
 752     {
 753         DSPThread* thread = static_cast<DSPThread*>(arg);
 754
 755         AVOIDDENORMALS;
 756
 757         // One "dummy" cycle to setup thread
 758         if (thread->fRealTime) {
 759             thread->Run();
 760             SetRealTime();
 761         }
 762
 763         while (true) {
 764             thread->Run();
 765         }
 766
 767         return NULL;
 768     }
 769
 770     int Start(bool realtime)
 771     {
 772         pthread_attr_t attributes;
 773         struct sched_param rt_param;
 774         pthread_attr_init(&attributes);
 775
 776         int priority = 60; // TODO
 777         int res;
 778
 779         if (realtime) {
 780             fRealTime = true;
 781         }else {
 782             fRealTime = getenv("OMP_REALTIME") ? strtol(getenv("OMP_REALTIME"), NULL, 10) : true;
 783         }
 784
 785         if ((res = pthread_attr_setdetachstate(&attributes, PTHREAD_CREATE_JOINABLE))) {
 786             printf("Cannot request joinable thread creation for real-time thread res = %d err = %s\n", res, strerror(errno));
 787             return -1;
 788         }
 789
 790         if ((res = pthread_attr_setscope(&attributes, PTHREAD_SCOPE_SYSTEM))) {
 791             printf("Cannot set scheduling scope for real-time thread res = %d err = %s\n", res, strerror(errno));
 792             return -1;
 793         }
 794
 795         if (realtime) {
 796
 797             if ((res = pthread_attr_setinheritsched(&attributes, PTHREAD_EXPLICIT_SCHED))) {
 798                 printf("Cannot request explicit scheduling for RT thread res = %d err = %s\n", res, strerror(errno));
 799                 return -1;
 800             }
 801
 802             if ((res = pthread_attr_setschedpolicy(&attributes, JACK_SCHED_POLICY))) {
 803                 printf("Cannot set RR scheduling class for RT thread res = %d err = %s\n", res, strerror(errno));
 804                 return -1;
 805             }
 806
 807             memset(&rt_param, 0, sizeof(rt_param));
 808             rt_param.sched_priority = priority;
 809
 810             if ((res = pthread_attr_setschedparam(&attributes, &rt_param))) {
 811                 printf("Cannot set scheduling priority for RT thread res = %d err = %s\n", res, strerror(errno));
 812                 return -1;
 813             }
 814
 815         } else {
 816
 817             if ((res = pthread_attr_setinheritsched(&attributes, PTHREAD_INHERIT_SCHED))) {
 818                 printf("Cannot request explicit scheduling for RT thread res = %d err = %s\n", res, strerror(errno));
 819                 return -1;
 820             }
 821         }
 822
 823         if ((res = pthread_attr_setstacksize(&attributes, THREAD_STACK))) {
 824             printf("Cannot set thread stack size res = %d err = %s\n", res, strerror(errno));
 825             return -1;
 826         }
 827
 828         if ((res = pthread_create(&fThread, &attributes, ThreadHandler, this))) {
 829             printf("Cannot create thread res = %d err = %s\n", res, strerror(errno));
 830             return -1;
 831         }
 832
 833         pthread_attr_destroy(&attributes);
 834         return 0;
 835     }
 836
 837     void Signal(bool stop, Runnable* runnable)
 838     {
 839         fRunnable = runnable;
 840         sem_post(fSemaphore);
 841     }
 842
 843     void Stop()
 844     {
 845         CancelThread(fThread);
 846     }
 847
 848 };
 849
 850 DSPThreadPool::DSPThreadPool()
 851 {
 852     for (int i = 0; i < THREAD_POOL_SIZE; i++) {
 853         fThreadPool[i] = NULL;
 854     }
 855     fThreadCount = 0;
 856     fCurThreadCount = 0;
 857 }
 858
 859 DSPThreadPool::~DSPThreadPool()
 860 {
 861     StopAll();
 862
 863     for (int i = 0; i < fThreadCount; i++) {
 864         delete(fThreadPool[i]);
 865         fThreadPool[i] = NULL;
 866     }
 867
 868     fThreadCount = 0;
 869  }
 870
 871 void DSPThreadPool::StartAll(int num, bool realtime)
 872 {
 873     if (fThreadCount == 0) {  // Protection for multiple call...  (like LADSPA plug-ins in Ardour)
 874         for (int i = 0; i < num; i++) {
 875             fThreadPool[i] = new DSPThread(i, this);
 876             fThreadPool[i]->Start(realtime);
 877             fThreadCount++;
 878         }
 879     }
 880 }
 881
 882 void DSPThreadPool::StopAll()
 883 {
 884     for (int i = 0; i < fThreadCount; i++) {
 885         fThreadPool[i]->Stop();
 886     }
 887 }
 888
 889 void DSPThreadPool::SignalAll(int num, Runnable* runnable)
 890 {
 891     fCurThreadCount = num;
 892
 893     for (int i = 0; i < num; i++) {  // Important : use local num here...
 894         fThreadPool[i]->Signal(false, runnable);
 895     }
 896 }
 897
 898 void DSPThreadPool::SignalOne()
 899 {
 900     DEC_ATOMIC(&fCurThreadCount);
 901 }
 902
 903 bool DSPThreadPool::IsFinished()
 904 {
 905     return (fCurThreadCount == 0);
 906 }
 907
 908 DSPThreadPool* DSPThreadPool::Init()
 909 {
 910     if (gClientCount++ == 0 && !gThreadPool) {
 911         gThreadPool = new DSPThreadPool();
 912     }
 913     return gThreadPool;
 914 }
 915
 916 void DSPThreadPool::Destroy()
 917 {
 918     if (--gClientCount == 0 && gThreadPool) {
 919         delete gThreadPool;
 920         gThreadPool = NULL;
 921     }
 922 }
 923
 924 #ifndef PLUG_IN
 925
 926 // Globals
 927 TaskQueue* gTaskQueueList[THREAD_SIZE] = {0};
 928
 929 DSPThreadPool* gThreadPool = 0;
 930 int gClientCount = 0;
 931
 932 int clock_per_microsec = (getenv("CLOCKSPERSEC")
 933                 ? strtoll(getenv("CLOCKSPERSEC"), NULL, 10)
 934                 : DEFAULT_CLOCKSPERSEC) / 1000000;
 935
 936 UInt64  gMaxStealing = getenv("OMP_STEALING_DUR")
 937                 ? strtoll(getenv("OMP_STEALING_DUR"), NULL, 10) * clock_per_microsec
 938                 : MAX_STEAL_DUR * clock_per_microsec;
 939
 940 #endif
 941