Halide  12.0.1
Halide compiler and libraries
thread_pool_common.h
Go to the documentation of this file.
1 #define EXTENDED_DEBUG 0
2 
3 #if EXTENDED_DEBUG
4 // This code is currently setup for Linux debugging. Switch to using pthread_self on e.g. Mac OS X.
5 extern "C" int syscall(int);
6 
7 namespace {
8 int gettid() {
9 #ifdef BITS_32
10  return syscall(224);
11 #else
12  return syscall(186);
13 #endif
14 }
15 } // namespace
16 
17 #define log_message(stuff) print(nullptr) << gettid() << ": " << stuff << "\n"
18 #else
19 #define log_message(stuff)
20 #endif
21 
22 namespace Halide {
23 namespace Runtime {
24 namespace Internal {
25 
26 struct work {
28 
29  // If we come in to the task system via do_par_for we just have a
30  // halide_task_t, not a halide_loop_task_t.
32 
38 
39  void *user_context;
43  // which condition variable is the owner sleeping on. nullptr if it isn't sleeping.
45 
50  // Note that we don't release the semaphores already
51  // acquired. We never have two consumers contending
52  // over the same semaphore, so it's not helpful to do
53  // so.
54  return false;
55  }
56  }
57  // Future iterations of this task need to acquire the semaphores from scratch.
58  next_semaphore = 0;
59  return true;
60  }
61 
62  ALWAYS_INLINE bool running() const {
63  return task.extent || active_workers;
64  }
65 };
66 
67 #define MAX_THREADS 256
68 
69 WEAK int clamp_num_threads(int threads) {
70  if (threads > MAX_THREADS) {
71  threads = MAX_THREADS;
72  } else if (threads < 1) {
73  threads = 1;
74  }
75  return threads;
76 }
77 
79  int desired_num_threads = 0;
80  char *threads_str = getenv("HL_NUM_THREADS");
81  if (!threads_str) {
82  // Legacy name for HL_NUM_THREADS
83  threads_str = getenv("HL_NUMTHREADS");
84  }
85  if (threads_str) {
86  desired_num_threads = atoi(threads_str);
87  } else {
88  desired_num_threads = halide_host_cpu_count();
89  }
90  return desired_num_threads;
91 }
92 
93 // The work queue and thread pool is weak, so one big work queue is shared by all halide functions
94 struct work_queue_t {
95  // all fields are protected by this mutex.
97 
98  // The desired number threads doing work (HL_NUM_THREADS).
100 
101  // All fields after this must be zero in the initial state. See assert_zeroed
102  // Field serves both to mark the offset in struct and as layout padding.
104 
105  // Singly linked list for job stack
107 
108  // The number threads created
110 
111  // Workers sleep on one of two condition variables, to make it
112  // easier to wake up the right number if a small number of tasks
113  // are enqueued. There are A-team workers and B-team workers. The
114  // following variables track the current size and the desired size
115  // of the A team.
117 
118  // The condition variables that workers and owners sleep on. We
119  // may want to wake them up independently. Any code that may
120  // invalidate any of the reasons a worker or owner may have slept
121  // must signal or broadcast the appropriate condition variable.
123 
124  // The number of sleeping workers and owners. An over-estimate - a
125  // waking-up thread may not have decremented this yet.
127 
128  // Keep track of threads so they can be joined at shutdown
129  halide_thread *threads[MAX_THREADS];
130 
131  // Global flags indicating the threadpool should shut down, and
132  // whether the thread pool has been initialized.
134 
135  // The number of threads that are currently commited to possibly block
136  // via outstanding jobs queued or being actively worked on. Used to limit
137  // the number of iterations of parallel for loops that are invoked so as
138  // to prevent deadlock due to oversubscription of threads.
140 
141  ALWAYS_INLINE bool running() const {
142  return !shutdown;
143  }
144 
145  // Used to check initial state is correct.
147  // Assert that all fields except the mutex and desired hreads count are zeroed.
148  const char *bytes = ((const char *)&this->zero_marker);
149  const char *limit = ((const char *)this) + sizeof(work_queue_t);
150  while (bytes < limit && *bytes == 0) {
151  bytes++;
152  }
153  halide_assert(nullptr, bytes == limit && "Logic error in thread pool work queue initialization.\n");
154  }
155 
156  // Return the work queue to initial state. Must be called while locked
157  // and queue will remain locked.
159  // Ensure all fields except the mutex and desired hreads count are zeroed.
160  char *bytes = ((char *)&this->zero_marker);
161  char *limit = ((char *)this) + sizeof(work_queue_t);
162  memset(bytes, 0, limit - bytes);
163  }
164 };
165 
167 
168 #if EXTENDED_DEBUG
169 WEAK void print_job(work *job, const char *indent, const char *prefix = nullptr) {
170  if (prefix == nullptr) {
171  prefix = indent;
172  }
173  const char *name = job->task.name ? job->task.name : "<no name>";
174  const char *parent_name = job->parent_job ? (job->parent_job->task.name ? job->parent_job->task.name : "<no name>") : "<no parent job>";
175  log_message(prefix << name << "[" << job << "] serial: " << job->task.serial << " active_workers: " << job->active_workers << " min: " << job->task.min << " extent: " << job->task.extent << " siblings: " << job->siblings << " sibling count: " << job->sibling_count << " min_threads " << job->task.min_threads << " next_sempaphore: " << job->next_semaphore << " threads_reserved: " << job->threads_reserved << " parent_job: " << parent_name << "[" << job->parent_job << "]");
176  for (int i = 0; i < job->task.num_semaphores; i++) {
177  log_message(indent << " semaphore " << (void *)job->task.semaphores[i].semaphore << " count " << job->task.semaphores[i].count << " val " << *(int *)job->task.semaphores[i].semaphore);
178  }
179 }
180 
181 WEAK void dump_job_state() {
182  log_message("Dumping job state, jobs in queue:");
183  work *job = work_queue.jobs;
184  while (job != nullptr) {
185  print_job(job, " ");
186  job = job->next_job;
187  }
188  log_message("Done dumping job state.");
189 }
190 #else
191 #define print_job(job, indent, prefix)
192 #define dump_job_state()
193 #endif
194 
195 WEAK void worker_thread(void *);
196 
198  int spin_count = 0;
199  const int max_spin_count = 40;
200 
201  while (owned_job ? owned_job->running() : !work_queue.shutdown) {
202  work *job = work_queue.jobs;
203  work **prev_ptr = &work_queue.jobs;
204 
205  if (owned_job) {
206  if (owned_job->exit_status != 0) {
207  if (owned_job->active_workers == 0) {
208  while (job != owned_job) {
209  prev_ptr = &job->next_job;
210  job = job->next_job;
211  }
212  *prev_ptr = job->next_job;
213  job->task.extent = 0;
214  continue; // So loop exit is always in the same place.
215  }
216  } else if (owned_job->parent_job && owned_job->parent_job->exit_status != 0) {
217  owned_job->exit_status = owned_job->parent_job->exit_status;
218  // The wakeup can likely be only done under certain conditions, but it is only happening
219  // in when an error has already occured and it seems more important to ensure reliable
220  // termination than to optimize this path.
222  continue;
223  }
224  }
225 
226  dump_job_state();
227 
228  // Find a job to run, prefering things near the top of the stack.
229  while (job) {
230  print_job(job, "", "Considering job ");
231  // Only schedule tasks with enough free worker threads
232  // around to complete. They may get stolen later, but only
233  // by tasks which can themselves use them to complete
234  // work, so forward progress is made.
235  bool enough_threads;
236 
237  work *parent_job = job->parent_job;
238 
239  int threads_available;
240  if (parent_job == nullptr) {
241  // The + 1 is because work_queue.threads_created does not include the main thread.
242  threads_available = (work_queue.threads_created + 1) - work_queue.threads_reserved;
243  } else {
244  if (parent_job->active_workers == 0) {
245  threads_available = parent_job->task.min_threads - parent_job->threads_reserved;
246  } else {
247  threads_available = parent_job->active_workers * parent_job->task.min_threads - parent_job->threads_reserved;
248  }
249  }
250  enough_threads = threads_available >= job->task.min_threads;
251 
252  if (!enough_threads) {
253 
254  log_message("Not enough threads for job " << job->task.name << " available: " << threads_available << " min_threads: " << job->task.min_threads);
255  }
256  bool can_use_this_thread_stack = !owned_job || (job->siblings == owned_job->siblings) || job->task.min_threads == 0;
257  if (!can_use_this_thread_stack) {
258  log_message("Cannot run job " << job->task.name << " on this thread.");
259  }
260  bool can_add_worker = (!job->task.serial || (job->active_workers == 0));
261  if (!can_add_worker) {
262  log_message("Cannot add worker to job " << job->task.name);
263  }
264 
265  if (enough_threads && can_use_this_thread_stack && can_add_worker) {
266  if (job->make_runnable()) {
267  break;
268  } else {
269  log_message("Cannot acquire semaphores for " << job->task.name);
270  }
271  }
272  prev_ptr = &(job->next_job);
273  job = job->next_job;
274  }
275 
276  if (!job) {
277  // There is no runnable job. Go to sleep.
278  if (owned_job) {
279  if (spin_count++ < max_spin_count) {
280  // Give the workers a chance to finish up before sleeping
284  } else {
286  owned_job->owner_is_sleeping = true;
288  owned_job->owner_is_sleeping = false;
290  }
291  } else {
294  // Transition to B team
298  } else if (spin_count++ < max_spin_count) {
299  // Spin waiting for new work
303  } else {
305  }
307  }
308  continue;
309  } else {
310  spin_count = 0;
311  }
312 
313  log_message("Working on job " << job->task.name);
314 
315  // Increment the active_worker count so that other threads
316  // are aware that this job is still in progress even
317  // though there are no outstanding tasks for it.
318  job->active_workers++;
319 
320  if (job->parent_job == nullptr) {
322  log_message("Reserved " << job->task.min_threads << " on work queue for " << job->task.name << " giving " << work_queue.threads_reserved << " of " << work_queue.threads_created + 1);
323  } else {
325  log_message("Reserved " << job->task.min_threads << " on " << job->parent_job->task.name << " for " << job->task.name << " giving " << job->parent_job->threads_reserved << " of " << job->parent_job->task.min_threads);
326  }
327 
328  int result = 0;
329 
330  if (job->task.serial) {
331  // Remove it from the stack while we work on it
332  *prev_ptr = job->next_job;
333 
334  // Release the lock and do the task.
336  int total_iters = 0;
337  int iters = 1;
338  while (result == 0) {
339  // Claim as many iterations as possible
340  while ((job->task.extent - total_iters) > iters &&
341  job->make_runnable()) {
342  iters++;
343  }
344  if (iters == 0) {
345  break;
346  }
347 
348  // Do them
349  result = halide_do_loop_task(job->user_context, job->task.fn,
350  job->task.min + total_iters, iters,
351  job->task.closure, job);
352  total_iters += iters;
353  iters = 0;
354  }
356 
357  job->task.min += total_iters;
358  job->task.extent -= total_iters;
359 
360  // Put it back on the job stack, if it hasn't failed.
361  if (result != 0) {
362  job->task.extent = 0; // Force job to be finished.
363  } else if (job->task.extent > 0) {
364  job->next_job = work_queue.jobs;
365  work_queue.jobs = job;
366  }
367  } else {
368  // Claim a task from it.
369  work myjob = *job;
370  job->task.min++;
371  job->task.extent--;
372 
373  // If there were no more tasks pending for this job, remove it
374  // from the stack.
375  if (job->task.extent == 0) {
376  *prev_ptr = job->next_job;
377  }
378 
379  // Release the lock and do the task.
381  if (myjob.task_fn) {
382  result = halide_do_task(myjob.user_context, myjob.task_fn,
383  myjob.task.min, myjob.task.closure);
384  } else {
385  result = halide_do_loop_task(myjob.user_context, myjob.task.fn,
386  myjob.task.min, 1,
387  myjob.task.closure, job);
388  }
390  }
391 
392  if (result != 0) {
393  log_message("Saw thread pool saw error from task: " << result);
394  }
395 
396  bool wake_owners = false;
397 
398  // If this task failed, set the exit status on the job.
399  if (result != 0) {
400  job->exit_status = result;
401  // Mark all siblings as also failed.
402  for (int i = 0; i < job->sibling_count; i++) {
403  log_message("Marking " << job->sibling_count << " siblings ");
404  if (job->siblings[i].exit_status == 0) {
405  job->siblings[i].exit_status = result;
406  wake_owners |= (job->active_workers == 0 && job->siblings[i].owner_is_sleeping);
407  }
408  log_message("Done marking siblings.");
409  }
410  }
411 
412  if (job->parent_job == nullptr) {
414  log_message("Returned " << job->task.min_threads << " to work queue for " << job->task.name << " giving " << work_queue.threads_reserved << " of " << work_queue.threads_created + 1);
415  } else {
417  log_message("Returned " << job->task.min_threads << " to " << job->parent_job->task.name << " for " << job->task.name << " giving " << job->parent_job->threads_reserved << " of " << job->parent_job->task.min_threads);
418  }
419 
420  // We are no longer active on this job
421  job->active_workers--;
422 
423  log_message("Done working on job " << job->task.name);
424 
425  if (wake_owners ||
426  (job->active_workers == 0 && (job->task.extent == 0 || job->exit_status != 0) && job->owner_is_sleeping)) {
427  // The job is done or some owned job failed via sibling linkage. Wake up the owner.
429  }
430  }
431 }
432 
433 WEAK void worker_thread(void *arg) {
437 }
438 
439 WEAK void enqueue_work_already_locked(int num_jobs, work *jobs, work *task_parent) {
440  if (!work_queue.initialized) {
442 
443  // Compute the desired number of threads to use. Other code
444  // can also mess with this value, but only when the work queue
445  // is locked.
448  }
450  work_queue.initialized = true;
451  }
452 
453  // Gather some information about the work.
454 
455  // Some tasks require a minimum number of threads to make forward
456  // progress. Also assume the blocking tasks need to run concurrently.
457  int min_threads = 0;
458 
459  // Count how many workers to wake. Start at -1 because this thread
460  // will contribute.
461  int workers_to_wake = -1;
462 
463  // Could stalled owners of other tasks conceivably help with one
464  // of these jobs.
465  bool stealable_jobs = false;
466 
467  bool job_has_acquires = false;
468  bool job_may_block = false;
469  for (int i = 0; i < num_jobs; i++) {
470  if (jobs[i].task.min_threads == 0) {
471  stealable_jobs = true;
472  } else {
473  job_may_block = true;
474  min_threads += jobs[i].task.min_threads;
475  }
476  if (jobs[i].task.num_semaphores != 0) {
477  job_has_acquires = true;
478  }
479 
480  if (jobs[i].task.serial) {
481  workers_to_wake++;
482  } else {
483  workers_to_wake += jobs[i].task.extent;
484  }
485  }
486 
487  if (task_parent == nullptr) {
488  // This is here because some top-level jobs may block, but are not accounted for
489  // in any enclosing min_threads count. In order to handle extern stages and such
490  // correctly, we likely need to make the total min_threads for an invocation of
491  // a pipeline a property of the entire thing. This approach works because we use
492  // the increased min_threads count to increase the size of the thread pool. It should
493  // even be safe against reservation races because this is happening under the work
494  // queue lock and that lock will be held into running the job. However that's many
495  // lines of code from here to there and it is not guaranteed this will be the first
496  // job run.
497  if (job_has_acquires || job_may_block) {
498  log_message("enqueue_work_already_locked adding one to min_threads.");
499  min_threads += 1;
500  }
501 
502  // Spawn more threads if necessary.
505  (work_queue.threads_created + 1) - work_queue.threads_reserved < min_threads)) {
506  // We might need to make some new threads, if work_queue.desired_threads_working has
507  // increased, or if there aren't enough threads to complete this new task.
511  }
512  log_message("enqueue_work_already_locked top level job " << jobs[0].task.name << " with min_threads " << min_threads << " work_queue.threads_created " << work_queue.threads_created << " work_queue.threads_reserved " << work_queue.threads_reserved);
513  if (job_has_acquires || job_may_block) {
515  }
516  } else {
517  log_message("enqueue_work_already_locked job " << jobs[0].task.name << " with min_threads " << min_threads << " task_parent " << task_parent->task.name << " task_parent->task.min_threads " << task_parent->task.min_threads << " task_parent->threads_reserved " << task_parent->threads_reserved);
518  halide_assert(nullptr, (min_threads <= ((task_parent->task.min_threads * task_parent->active_workers) -
519  task_parent->threads_reserved)) &&
520  "Logic error: thread over commit.\n");
521  if (job_has_acquires || job_may_block) {
522  task_parent->threads_reserved++;
523  }
524  }
525 
526  // Push the jobs onto the stack.
527  for (int i = num_jobs - 1; i >= 0; i--) {
528  // We could bubble it downwards based on some heuristics, but
529  // it's not strictly necessary to do so.
530  jobs[i].next_job = work_queue.jobs;
531  jobs[i].siblings = &jobs[0];
532  jobs[i].sibling_count = num_jobs;
533  jobs[i].threads_reserved = 0;
534  work_queue.jobs = jobs + i;
535  }
536 
537  bool nested_parallelism =
540 
541  // Wake up an appropriate number of threads
542  if (nested_parallelism || workers_to_wake > work_queue.workers_sleeping) {
543  // If there's nested parallelism going on, we just wake up
544  // everyone. TODO: make this more precise.
546  } else {
547  work_queue.target_a_team_size = workers_to_wake;
548  }
549 
553  if (stealable_jobs) {
555  }
556  }
557 
558  if (job_has_acquires || job_may_block) {
559  if (task_parent != nullptr) {
560  task_parent->threads_reserved--;
561  } else {
563  }
564  }
565 }
566 
574 
575 } // namespace Internal
576 } // namespace Runtime
577 } // namespace Halide
578 
579 using namespace Halide::Runtime::Internal;
580 
581 extern "C" {
582 
583 namespace {
584 WEAK __attribute__((destructor)) void halide_thread_pool_cleanup() {
586 }
587 } // namespace
588 
590  uint8_t *closure) {
591  return f(user_context, idx, closure);
592 }
593 
595  int min, int extent, uint8_t *closure,
596  void *task_parent) {
597  return f(user_context, min, extent, closure, task_parent);
598 }
599 
601  int min, int size, uint8_t *closure) {
602  if (size <= 0) {
603  return 0;
604  }
605 
606  work job;
607  job.task.fn = nullptr;
608  job.task.min = min;
609  job.task.extent = size;
610  job.task.serial = false;
611  job.task.semaphores = nullptr;
612  job.task.num_semaphores = 0;
613  job.task.closure = closure;
614  job.task.min_threads = 0;
615  job.task.name = nullptr;
616  job.task_fn = f;
618  job.exit_status = 0;
619  job.active_workers = 0;
620  job.next_semaphore = 0;
621  job.owner_is_sleeping = false;
622  job.siblings = &job; // guarantees no other job points to the same siblings.
623  job.sibling_count = 0;
624  job.parent_job = nullptr;
626  enqueue_work_already_locked(1, &job, nullptr);
629  return job.exit_status;
630 }
631 
633  struct halide_parallel_task_t *tasks,
634  void *task_parent) {
635  work *jobs = (work *)__builtin_alloca(sizeof(work) * num_tasks);
636 
637  for (int i = 0; i < num_tasks; i++) {
638  if (tasks->extent <= 0) {
639  // Skip extent zero jobs
640  num_tasks--;
641  continue;
642  }
643  jobs[i].task = *tasks++;
644  jobs[i].task_fn = nullptr;
645  jobs[i].user_context = user_context;
646  jobs[i].exit_status = 0;
647  jobs[i].active_workers = 0;
648  jobs[i].next_semaphore = 0;
649  jobs[i].owner_is_sleeping = false;
650  jobs[i].parent_job = (work *)task_parent;
651  }
652 
653  if (num_tasks == 0) {
654  return 0;
655  }
656 
658  enqueue_work_already_locked(num_tasks, jobs, (work *)task_parent);
659  int exit_status = 0;
660  for (int i = 0; i < num_tasks; i++) {
661  // It doesn't matter what order we join the tasks in, because
662  // we'll happily assist with siblings too.
664  if (jobs[i].exit_status != 0) {
665  exit_status = jobs[i].exit_status;
666  }
667  }
669  return exit_status;
670 }
671 
673  if (n < 0) {
674  halide_error(nullptr, "halide_set_num_threads: must be >= 0.");
675  }
676  // Don't make this an atomic swap - we don't want to be changing
677  // the desired number of threads while another thread is in the
678  // middle of a sequence of non-atomic operations.
680  if (n == 0) {
682  }
686  return old;
687 }
688 
690  if (work_queue.initialized) {
691  // Wake everyone up and tell them the party's over and it's time
692  // to go home
694 
695  work_queue.shutdown = true;
700 
701  // Wait until they leave
702  for (int i = 0; i < work_queue.threads_created; i++) {
704  }
705 
706  // Tidy up
707  work_queue.reset();
708  }
709 }
710 
712  int value;
713 };
714 
717  Halide::Runtime::Internal::Synchronization::atomic_store_release(&sem->value, &n);
718  return n;
719 }
720 
723  int old_val = Halide::Runtime::Internal::Synchronization::atomic_fetch_add_acquire_release(&sem->value, n);
724  // TODO(abadams|zvookin): Is this correct if an acquire can be for say count of 2 and the releases are 1 each?
725  if (old_val == 0 && n != 0) { // Don't wake if nothing released.
726  // We may have just made a job runnable
731  }
732  return old_val + n;
733 }
734 
736  if (n == 0) {
737  return true;
738  }
740  // Decrement and get new value
741  int expected;
742  int desired;
743  Halide::Runtime::Internal::Synchronization::atomic_load_acquire(&sem->value, &expected);
744  do {
745  desired = expected - n;
746  } while (desired >= 0 &&
747  !Halide::Runtime::Internal::Synchronization::atomic_cas_weak_relacq_relaxed(&sem->value, &expected, &desired));
748  return desired >= 0;
749 }
750 
753  custom_do_task = f;
754  return result;
755 }
756 
760  return result;
761 }
762 
765  custom_do_par_for = f;
766  return result;
767 }
768 
770  halide_do_par_for_t do_par_for,
771  halide_do_task_t do_task,
772  halide_do_loop_task_t do_loop_task,
773  halide_do_parallel_tasks_t do_parallel_tasks,
774  halide_semaphore_init_t semaphore_init,
775  halide_semaphore_try_acquire_t semaphore_try_acquire,
776  halide_semaphore_release_t semaphore_release) {
777 
778  custom_do_par_for = do_par_for;
779  custom_do_task = do_task;
780  custom_do_loop_task = do_loop_task;
781  custom_do_parallel_tasks = do_parallel_tasks;
782  custom_semaphore_init = semaphore_init;
783  custom_semaphore_try_acquire = semaphore_try_acquire;
784  custom_semaphore_release = semaphore_release;
785 }
786 
788  uint8_t *closure) {
789  return (*custom_do_task)(user_context, f, idx, closure);
790 }
791 
793  int min, int size, uint8_t *closure) {
794  return (*custom_do_par_for)(user_context, f, min, size, closure);
795 }
796 
798  int min, int size, uint8_t *closure, void *task_parent) {
799  return custom_do_loop_task(user_context, f, min, size, closure, task_parent);
800 }
801 
802 WEAK int halide_do_parallel_tasks(void *user_context, int num_tasks,
803  struct halide_parallel_task_t *tasks,
804  void *task_parent) {
805  return custom_do_parallel_tasks(user_context, num_tasks, tasks, task_parent);
806 }
807 
808 WEAK int halide_semaphore_init(struct halide_semaphore_t *sema, int count) {
809  return custom_semaphore_init(sema, count);
810 }
811 
812 WEAK int halide_semaphore_release(struct halide_semaphore_t *sema, int count) {
813  return custom_semaphore_release(sema, count);
814 }
815 
816 WEAK bool halide_semaphore_try_acquire(struct halide_semaphore_t *sema, int count) {
817  return custom_semaphore_try_acquire(sema, count);
818 }
819 }
int(* halide_semaphore_release_t)(struct halide_semaphore_t *, int)
void halide_cond_wait(struct halide_cond *cond, struct halide_mutex *mutex)
int(* halide_do_par_for_t)(void *, halide_task_t, int, int, uint8_t *)
Set a custom method for performing a parallel for loop.
int(* halide_task_t)(void *user_context, int task_number, uint8_t *closure)
Define halide_do_par_for to replace the default thread pool implementation.
void halide_mutex_lock(struct halide_mutex *mutex)
A basic set of mutex and condition variable functions, which call platform specific code for mutual e...
void halide_mutex_unlock(struct halide_mutex *mutex)
struct halide_thread * halide_spawn_thread(void(*f)(void *), void *closure)
Spawn a thread.
int(* halide_do_loop_task_t)(void *, halide_loop_task_t, int, int, uint8_t *, void *)
The version of do_task called for loop tasks.
bool(* halide_semaphore_try_acquire_t)(struct halide_semaphore_t *, int)
int(* halide_loop_task_t)(void *user_context, int min, int extent, uint8_t *closure, void *task_parent)
A task representing a serial for loop evaluated over some range.
void halide_join_thread(struct halide_thread *)
Join a thread.
void halide_cond_broadcast(struct halide_cond *cond)
int(* halide_do_task_t)(void *, halide_task_t, int, uint8_t *)
If you use the default do_par_for, you can still set a custom handler to perform each individual task...
int(* halide_semaphore_init_t)(struct halide_semaphore_t *, int)
void halide_error(void *user_context, const char *)
Halide calls this function on runtime errors (for example bounds checking failures).
int(* halide_do_parallel_tasks_t)(void *, int, struct halide_parallel_task_t *, void *task_parent)
Provide an entire custom tasking runtime via function pointers.
WEAK halide_semaphore_release_t custom_semaphore_release
WEAK halide_semaphore_init_t custom_semaphore_init
WEAK halide_do_task_t custom_do_task
WEAK halide_do_par_for_t custom_do_par_for
WEAK void worker_thread(void *)
WEAK int clamp_num_threads(int threads)
WEAK void enqueue_work_already_locked(int num_jobs, work *jobs, work *task_parent)
WEAK halide_do_parallel_tasks_t custom_do_parallel_tasks
WEAK void worker_thread_already_locked(work *owned_job)
WEAK halide_do_loop_task_t custom_do_loop_task
WEAK halide_semaphore_try_acquire_t custom_semaphore_try_acquire
This file defines the class FunctionDAG, which is our representation of a Halide pipeline,...
Expr min(const FuncRef &a, const FuncRef &b)
Explicit overloads of min and max for FuncRef.
Definition: Func.h:578
void * user_context
Definition: printer.h:33
WEAK int halide_host_cpu_count()
int atoi(const char *)
unsigned __INT8_TYPE__ uint8_t
void * memset(void *s, int val, size_t n)
void halide_thread_yield()
char * getenv(const char *)
#define ALWAYS_INLINE
#define halide_assert(user_context, cond)
#define WEAK
halide_thread * threads[MAX_THREADS]
ALWAYS_INLINE void assert_zeroed() const
ALWAYS_INLINE bool running() const
ALWAYS_INLINE bool make_runnable()
Cross platform condition variable.
Cross-platform mutex.
A parallel task to be passed to halide_do_parallel_tasks.
struct halide_semaphore_acquire_t * semaphores
halide_loop_task_t fn
struct halide_semaphore_t * semaphore
An opaque struct representing a semaphore.
WEAK void halide_set_custom_parallel_runtime(halide_do_par_for_t do_par_for, halide_do_task_t do_task, halide_do_loop_task_t do_loop_task, halide_do_parallel_tasks_t do_parallel_tasks, halide_semaphore_init_t semaphore_init, halide_semaphore_try_acquire_t semaphore_try_acquire, halide_semaphore_release_t semaphore_release)
WEAK int halide_default_semaphore_release(halide_semaphore_t *s, int n)
WEAK halide_do_task_t halide_set_custom_do_task(halide_do_task_t f)
#define dump_job_state()
WEAK bool halide_default_semaphore_try_acquire(halide_semaphore_t *s, int n)
WEAK bool halide_semaphore_try_acquire(struct halide_semaphore_t *sema, int count)
WEAK int halide_default_do_parallel_tasks(void *user_context, int num_tasks, struct halide_parallel_task_t *tasks, void *task_parent)
WEAK halide_do_loop_task_t halide_set_custom_do_loop_task(halide_do_loop_task_t f)
WEAK int halide_do_par_for(void *user_context, halide_task_t f, int min, int size, uint8_t *closure)
WEAK int halide_semaphore_init(struct halide_semaphore_t *sema, int count)
WEAK int halide_default_do_loop_task(void *user_context, halide_loop_task_t f, int min, int extent, uint8_t *closure, void *task_parent)
#define MAX_THREADS
#define log_message(stuff)
WEAK halide_do_par_for_t halide_set_custom_do_par_for(halide_do_par_for_t f)
#define print_job(job, indent, prefix)
WEAK int halide_do_loop_task(void *user_context, halide_loop_task_t f, int min, int size, uint8_t *closure, void *task_parent)
WEAK int halide_do_parallel_tasks(void *user_context, int num_tasks, struct halide_parallel_task_t *tasks, void *task_parent)
Enqueue some number of the tasks described above and wait for them to complete.
WEAK void halide_shutdown_thread_pool()
WEAK int halide_default_semaphore_init(halide_semaphore_t *s, int n)
WEAK int halide_default_do_par_for(void *user_context, halide_task_t f, int min, int size, uint8_t *closure)
The default versions of the parallel runtime functions.
WEAK int halide_do_task(void *user_context, halide_task_t f, int idx, uint8_t *closure)
WEAK int halide_default_do_task(void *user_context, halide_task_t f, int idx, uint8_t *closure)
WEAK int halide_set_num_threads(int n)
Set the number of threads used by Halide's thread pool.
WEAK int halide_semaphore_release(struct halide_semaphore_t *sema, int count)