[OpenMP] Adding a throttling threshold to bound dependent tasking memory footprint

[rpereira-dev]

Summary

From "5. Discussions - Task Throttling" of this paper

Task Throttling is a runtime mechanism to reduce tasking operational and memory overheads.
Once at threshold is reached, producer threads stop producing and start consuming tasks instead.

LLVM runtime implements a threshold on (1) the number of ready-tasks per thread.
It was implemented in the context of independent tasking (OpenMP-3.0, 2008), with a motivation being to provide a memory consumption bound.
In the context of dependent tasking (OpenMP-4.0, 2013), this threshold is not sufficient to bound memory consumption: as many (not ready) successor tasks may be created.

This patch introduces a new threshold: (2) the total number of tasks and an environment variable KMP_TASK_MAXIMUM to configure its value (set to 65,536); providing strong warranties on memory consumption.
The patch also adds the KMP_TASK_MAXIMUM_READY to configure the threshold (1) (set to 256).

Few points that came to my mind that may need discussions:

• a) thresholds default values
• b) induced overheads: an atomic variable is written by any threads on each task allocation/deallocation - maybe a compile-time option could be added to disabled (2)-throttling entirely preserving the current behavior

If other points come to your mind, please let me know

[tianshilei1992]

Can you update the patch with context?

[rpereira-dev]

Added context with the patch

[rpereira-dev]

Few issues in this patch, fixing it, please wait

[rpereira-dev]

• Fixed tests comment that got formatted by previous git clang-format
• Fixed throttling on the maximum number of tasks per thread; previous version would crash if this was larger than the original queue size: queues are now resizing

openmp/runtime/src/kmp.h

typedef enum kmp_tasking_mode {
   tskm_extra_barrier = 1,
   tskm_task_teams = 2,
   tskm_max = 2
 } kmp_tasking_mode_t;
 
 extern kmp_tasking_mode_t
     __kmp_tasking_mode; /* determines how/when to execute tasks */
 extern int __kmp_task_stealing_constraint;
+extern std::atomic<kmp_int32> __kmp_n_tasks_in_flight;
 extern int __kmp_enable_task_throttling;
+extern kmp_int32 __kmp_task_maximum;
+extern kmp_int32 __kmp_task_maximum_ready_per_thread;
+
 extern kmp_int32 __kmp_default_device; // Set via OMP_DEFAULT_DEVICE if
 // specified, defaults to 0 otherwise
 // Set via OMP_MAX_TASK_PRIORITY if specified, defaults to 0 otherwise
 extern kmp_int32 __kmp_max_task_priority;
 // Set via KMP_TASKLOOP_MIN_TASKS if specified, defaults to 0 otherwise
 extern kmp_uint64 __kmp_taskloop_min_tasks;
 
 /* NOTE: kmp_taskdata_t and kmp_task_t structures allocated in single block with

openmp/runtime/src/kmp_global.cpp

 /* This check ensures that the compiler is passing the correct data type for the
    flags formal parameter of the function kmpc_omp_task_alloc(). If the type is
    not a 4-byte type, then give an error message about a non-positive length
    array pointing here.  If that happens, the kmp_tasking_flags_t structure must
    be redefined to have exactly 32 bits. */
 KMP_BUILD_ASSERT(sizeof(kmp_tasking_flags_t) == 4);
 
 int __kmp_task_stealing_constraint = 1; /* Constrain task stealing by default */
-int __kmp_enable_task_throttling = 1;
 
+std::atomic<kmp_int32> __kmp_n_tasks_in_flight = 0; /* n° of tasks in flight */
+
+kmp_int32 __kmp_enable_task_throttling = 1; /* Serialize tasks once a threshold
+                                            is reached, such as the number of
+                                            ready tasks or the total number of
+                                            tasks */
+
+kmp_int32 __kmp_task_maximum = 65536; /* number of tasks threshold before
+                                         serializing */
+
+kmp_int32 __kmp_task_maximum_ready_per_thread = 256; /* number of ready tasks
+                                                        before serializing */
 #ifdef DEBUG_SUSPEND
 int __kmp_suspend_count = 0;
 #endif
 
 int __kmp_settings = FALSE;
 int __kmp_duplicate_library_ok = 0;
 #if USE_ITT_BUILD
 int __kmp_forkjoin_frames = 1;

openmp/runtime/src/kmp_settings.cpp

static void __kmp_stg_parse_task_throttling(char const *name, char const *value,
   __kmp_stg_parse_bool(name, value, &__kmp_enable_task_throttling);
 } // __kmp_stg_parse_task_throttling
 
 static void __kmp_stg_print_task_throttling(kmp_str_buf_t *buffer,
                                             char const *name, void *data) {
   __kmp_stg_print_bool(buffer, name, __kmp_enable_task_throttling);
 } // __kmp_stg_print_task_throttling
 
+// -----------------------------------------------------------------------------
+// KMP_TASK_MAXIMUM
+static void __kmp_stg_parse_task_maximum(char const *name, char const *value,
+                                         void *data) {
+  __kmp_stg_parse_int(name, value, 1, INT_MAX, &__kmp_task_maximum);
+} // __kmp_stg_parse_task_maximum
+
+static void __kmp_stg_print_task_maximum(kmp_str_buf_t *buffer,
+                                         char const *name, void *data) {
+  __kmp_stg_print_int(buffer, name, __kmp_task_maximum);
+} // __kmp_stg_print_task_maximum
+
+// -----------------------------------------------------------------------------
+// KMP_TASK_MAXIMUM_READY_PER_THREAD
+static void __kmp_stg_parse_task_maximum_ready_per_thread(char const *name,
+                                                          char const *value,
+                                                          void *data) {
+  __kmp_stg_parse_int(name, value, 1, INT_MAX,
+                      &__kmp_task_maximum_ready_per_thread);
+} // __kmp_stg_parse_task_maximum_ready_per_thread
+
+static void __kmp_stg_print_task_maximum_ready_per_thread(kmp_str_buf_t *buffer,
+                                                          char const *name,
+                                                          void *data) {
+  __kmp_stg_print_int(buffer, name, __kmp_task_maximum_ready_per_thread);
+} // __kmp_stg_print_task_maximum_ready_per_thread
+
 #if KMP_HAVE_MWAIT || KMP_HAVE_UMWAIT
 // -----------------------------------------------------------------------------
 // KMP_USER_LEVEL_MWAIT
 
 static void __kmp_stg_parse_user_level_mwait(char const *name,
                                              char const *value, void *data) {
   __kmp_stg_parse_bool(name, value, &__kmp_user_level_mwait);
 } // __kmp_stg_parse_user_level_mwait
#if USE_ITT_BUILD
     {"KMP_FORKJOIN_FRAMES", __kmp_stg_parse_forkjoin_frames,
      __kmp_stg_print_forkjoin_frames, NULL, 0, 0},
     {"KMP_FORKJOIN_FRAMES_MODE", __kmp_stg_parse_forkjoin_frames_mode,
      __kmp_stg_print_forkjoin_frames_mode, NULL, 0, 0},
 #endif
     {"KMP_ENABLE_TASK_THROTTLING", __kmp_stg_parse_task_throttling,
      __kmp_stg_print_task_throttling, NULL, 0, 0},
 
+    {"KMP_TASK_MAXIMUM", __kmp_stg_parse_task_maximum,
+     __kmp_stg_print_task_maximum, NULL, 0, 0},
+
+    {"KMP_TASK_MAXIMUM_READY_PER_THREAD",
+     __kmp_stg_parse_task_maximum_ready_per_thread,
+     __kmp_stg_print_task_maximum_ready_per_thread, NULL, 0, 0},
+
     {"OMP_DISPLAY_ENV", __kmp_stg_parse_omp_display_env,
      __kmp_stg_print_omp_display_env, NULL, 0, 0},
     {"OMP_CANCELLATION", __kmp_stg_parse_omp_cancellation,
      __kmp_stg_print_omp_cancellation, NULL, 0, 0},
     {"OMP_ALLOCATOR", __kmp_stg_parse_allocator, __kmp_stg_print_allocator,
      NULL, 0, 0},
     {"LIBOMP_USE_HIDDEN_HELPER_TASK", __kmp_stg_parse_use_hidden_helper,
      __kmp_stg_print_use_hidden_helper, NULL, 0, 0},
     {"LIBOMP_NUM_HIDDEN_HELPER_THREADS",
      __kmp_stg_parse_num_hidden_helper_threads,
      __kmp_stg_print_num_hidden_helper_threads, NULL, 0, 0},
 #if OMPX_TASKGRAPH
     {"KMP_MAX_TDGS", __kmp_stg_parse_max_tdgs, __kmp_std_print_max_tdgs, NULL,
      0, 0},
-    {"KMP_TDG_DOT", __kmp_stg_parse_tdg_dot, __kmp_stg_print_tdg_dot, NULL, 0, 0},
+    {"KMP_TDG_DOT", __kmp_stg_parse_tdg_dot, __kmp_stg_print_tdg_dot, NULL, 0,
+     0},
 #endif
 
 #if OMPT_SUPPORT
     {"OMP_TOOL", __kmp_stg_parse_omp_tool, __kmp_stg_print_omp_tool, NULL, 0,
      0},
     {"OMP_TOOL_LIBRARIES", __kmp_stg_parse_omp_tool_libraries,
      __kmp_stg_print_omp_tool_libraries, NULL, 0, 0},
     {"OMP_TOOL_VERBOSE_INIT", __kmp_stg_parse_omp_tool_verbose_init,

openmp/runtime/src/kmp_tasking.cpp

if (lst->priority == pri) {
       thread_data = __kmp_get_priority_deque_data(task_team, pri);
       __kmp_release_bootstrap_lock(&task_team->tt.tt_task_pri_lock);
     }
   }
   KMP_DEBUG_ASSERT(thread_data);
 
   __kmp_acquire_bootstrap_lock(&thread_data->td.td_deque_lock);
   // Check if deque is full
-  if (TCR_4(thread_data->td.td_deque_ntasks) >=
-      TASK_DEQUE_SIZE(thread_data->td)) {
-    if (__kmp_enable_task_throttling &&
-        __kmp_task_is_allowed(gtid, __kmp_task_stealing_constraint, taskdata,
+  if (__kmp_enable_task_throttling && TCR_4(thread_data->td.td_deque_ntasks) >=
+                                          __kmp_task_maximum_ready_per_thread) {
+    if (__kmp_task_is_allowed(gtid, __kmp_task_stealing_constraint, taskdata,
                               thread->th.th_current_task)) {
       __kmp_release_bootstrap_lock(&thread_data->td.td_deque_lock);
       KA_TRACE(20, ("__kmp_push_priority_task: T#%d deque is full; returning "
                     "TASK_NOT_PUSHED for task %p\n",
                     gtid, taskdata));
       return TASK_NOT_PUSHED;
     } else {
       // expand deque to push the task which is not allowed to execute
static kmp_int32 __kmp_push_task(kmp_int32 gtid, kmp_task_t *task) {
   // we don't need it either because we have initialized the dequeue for hidden
   // helper thread data.
   if (UNLIKELY(thread_data->td.td_deque == NULL)) {
     __kmp_alloc_task_deque(thread, thread_data);
   }
 
   int locked = 0;
   // Check if deque is full
-  if (TCR_4(thread_data->td.td_deque_ntasks) >=
-      TASK_DEQUE_SIZE(thread_data->td)) {
-    if (__kmp_enable_task_throttling &&
+  int requires_resize = TCR_4(thread_data->td.td_deque_ntasks) >=
+                        TASK_DEQUE_SIZE(thread_data->td);
+  int requires_throttling =
+      __kmp_enable_task_throttling && TCR_4(thread_data->td.td_deque_ntasks) >=
+                                          __kmp_task_maximum_ready_per_thread;
+  int thread_can_execute;
+  if (requires_resize || requires_throttling) {
+    thread_can_execute =
         __kmp_task_is_allowed(gtid, __kmp_task_stealing_constraint, taskdata,
-                              thread->th.th_current_task)) {
+                              thread->th.th_current_task);
+    if (requires_throttling && thread_can_execute) {
       KA_TRACE(20, ("__kmp_push_task: T#%d deque is full; returning "
                     "TASK_NOT_PUSHED for task %p\n",
                     gtid, taskdata));
       return TASK_NOT_PUSHED;
-    } else {
+    } else { /* maybe requires_resize */
       __kmp_acquire_bootstrap_lock(&thread_data->td.td_deque_lock);
       locked = 1;
-      if (TCR_4(thread_data->td.td_deque_ntasks) >=
-          TASK_DEQUE_SIZE(thread_data->td)) {
-        // expand deque to push the task which is not allowed to execute
+      requires_resize = TCR_4(thread_data->td.td_deque_ntasks) >=
+                        TASK_DEQUE_SIZE(thread_data->td);
+      if (requires_resize)
         __kmp_realloc_task_deque(thread, thread_data);
-      }
-    }
-  }
   // Lock the deque for the task push operation
   if (!locked) {
     __kmp_acquire_bootstrap_lock(&thread_data->td.td_deque_lock);
     // Need to recheck as we can get a proxy task from thread outside of OpenMP
-    if (TCR_4(thread_data->td.td_deque_ntasks) >=
-        TASK_DEQUE_SIZE(thread_data->td)) {
-      if (__kmp_enable_task_throttling &&
+    requires_resize = TCR_4(thread_data->td.td_deque_ntasks) >=
+                      TASK_DEQUE_SIZE(thread_data->td);
+    requires_throttling = __kmp_enable_task_throttling &&
+                          TCR_4(thread_data->td.td_deque_ntasks) >=
+                              __kmp_task_maximum_ready_per_thread;
+    if (requires_resize || requires_throttling) {
+      thread_can_execute =
           __kmp_task_is_allowed(gtid, __kmp_task_stealing_constraint, taskdata,
-                                thread->th.th_current_task)) {
+                                thread->th.th_current_task);
+      if (requires_throttling && thread_can_execute) {
         __kmp_release_bootstrap_lock(&thread_data->td.td_deque_lock);
         KA_TRACE(20, ("__kmp_push_task: T#%d deque is full on 2nd check; "
                       "returning TASK_NOT_PUSHED for task %p\n",
                       gtid, taskdata));
         return TASK_NOT_PUSHED;
-      } else {
+      } else { /* requires_resize */
         // expand deque to push the task which is not allowed to execute
         __kmp_realloc_task_deque(thread, thread_data);
       }
     }
   }
   // Must have room since no thread can add tasks but calling thread
   KMP_DEBUG_ASSERT(TCR_4(thread_data->td.td_deque_ntasks) <
                    TASK_DEQUE_SIZE(thread_data->td));
#if OMPX_TASKGRAPH
   if (!taskdata->is_taskgraph) {
 #endif
 // deallocate the taskdata and shared variable blocks associated with this task
 #if USE_FAST_MEMORY
   __kmp_fast_free(thread, taskdata);
 #else /* ! USE_FAST_MEMORY */
   __kmp_thread_free(thread, taskdata);
 #endif
+  --__kmp_n_tasks_in_flight;
 #if OMPX_TASKGRAPH
   } else {
     taskdata->td_flags.complete = 0;
     taskdata->td_flags.started = 0;
     taskdata->td_flags.freed = 0;
     taskdata->td_flags.executing = 0;
     taskdata->td_flags.task_serial =
         (taskdata->td_parent->td_flags.final ||
kmp_task_t *__kmp_task_alloc(ident_t *loc_ref, kmp_int32 gtid,
   kmp_info_t *thread = __kmp_threads[gtid];
   kmp_team_t *team = thread->th.th_team;
   kmp_taskdata_t *parent_task = thread->th.th_current_task;
   size_t shareds_offset;
 
   if (UNLIKELY(!TCR_4(__kmp_init_middle)))
     __kmp_middle_initialize();
 
+  // task throttling: to many tasks co-existing, emptying queue now
+  if (__kmp_enable_task_throttling)
+    while (TCR_4(__kmp_n_tasks_in_flight.load()) >= __kmp_task_maximum)
+      __kmpc_omp_taskyield(NULL, gtid, 0);
+
   if (flags->hidden_helper) {
     if (__kmp_enable_hidden_helper) {
       if (!TCR_4(__kmp_init_hidden_helper))
         __kmp_hidden_helper_initialize();
     } else {
       // If the hidden helper task is not enabled, reset the flag to FALSE.
       flags->hidden_helper = FALSE;
     }
kmp_task_t *__kmp_task_alloc(ident_t *loc_ref, kmp_int32 gtid,
   // Avoid double allocation here by combining shareds with taskdata
 #if USE_FAST_MEMORY
   taskdata = (kmp_taskdata_t *)__kmp_fast_allocate(thread, shareds_offset +
                                                                sizeof_shareds);
 #else /* ! USE_FAST_MEMORY */
   taskdata = (kmp_taskdata_t *)__kmp_thread_malloc(thread, shareds_offset +
                                                                sizeof_shareds);
 #endif /* USE_FAST_MEMORY */
+  ++__kmp_n_tasks_in_flight;
 
   task = KMP_TASKDATA_TO_TASK(taskdata);
 
 // Make sure task & taskdata are aligned appropriately
 #if KMP_ARCH_X86 || KMP_ARCH_PPC64 || !KMP_HAVE_QUAD
   KMP_DEBUG_ASSERT((((kmp_uintptr_t)taskdata) & (sizeof(double) - 1)) == 0);
   KMP_DEBUG_ASSERT((((kmp_uintptr_t)task) & (sizeof(double) - 1)) == 0);
 #else

openmp/runtime/test/tasking/omp_throttling_max.c

+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM=0      %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM=1      %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM=256    %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM=65536  %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM=100000 %libomp-run
+
+/**
+ *  This test ensures that task throttling on the maximum number of tasks
+ *  threshold works properly.
+ *
+ *  It creates 2 threads (1 producer, 1 consummer)
+ *  The producer infinitely create tasks 'T_i' until one executed
+ *  The consumer is blocked until the producer starts throttling
+ *  Executing any 'T_i' unblocks the consumer and stop the producer
+ *
+ *  The assertion tests ensures that the producer does not create more than the
+ *  total number of tasks provided by the programmer
+ */
+
+#include <assert.h>
+#include <omp.h>
+#include <stdlib.h>
+
+/* default value */
+#define MAX_TASKS_DEFAULT (65536)
+
+int main(void) {
+  /* maximum number of tasks in-flight */
+  char *max_tasks_str = getenv("KMP_TASK_MAXIMUM");
+  int max_tasks = max_tasks_str ? atoi(max_tasks_str) : MAX_TASKS_DEFAULT;
+  if (max_tasks <= 0)
+    max_tasks = 1;
+
+  /* check if throttling is enabled (it is by default) */
+  char *throttling_str = getenv("KMP_ENABLE_TASK_THROTTLING");
+  int throttling = throttling_str ? *throttling_str == '1' : 1;
+  assert(throttling);
+
+  volatile int done = 0;
+
+/* testing KMP_TASK_MAXIMUM */
+#pragma omp parallel num_threads(2) default(none)                              \
+    shared(max_tasks, throttling, done)
+  {
+    if (omp_get_thread_num() == 1)
+      while (!done)
+        ;
+
+#pragma omp master
+    {
+      int ntasks = 0;
+      while (!done) {
+#pragma omp task default(none) shared(done) depend(out : max_tasks, throttling)
+        done = 1;
+
+        assert(++ntasks <= max_tasks + 1);
+      }
+    }
+  }
+
+  return 0;
+}

openmp/runtime/test/tasking/omp_throttling_max_ready_per_thread.c

+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM_READY_PER_THREAD=0      %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM_READY_PER_THREAD=1      %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM_READY_PER_THREAD=256    %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM_READY_PER_THREAD=65536  %libomp-run
+// RUN: %libomp-compile && env OMP_NUM_THREADS=2 KMP_ENABLE_TASK_THROTTLING=1 KMP_TASK_MAXIMUM_READY_PER_THREAD=100000 %libomp-run
+
+/**
+ *  This test ensures that task throttling on the maximum number of ready tasks
+ *  per thread threshold works properly.
+ *
+ *  It creates 2 threads (1 producer, 1 consummer)
+ *  The producer infinitely create tasks 'T_i' until one executed
+ *  The consumer is blocked until the producer starts throttling
+ *  Executing any 'T_i' unblocks the consumer and stop the producer
+ *
+ *  The assertion tests ensures that the producer does not create more than the
+ *  total number of tasks provided by the programmer
+ */
+
+#include <assert.h>
+#include <omp.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#define MAX_TASKS_READY_DEFAULT (1 << 8)
+
+int main(void) {
+  /* maximum number of ready tasks in-flight */
+  char *max_tasks_ready_str = getenv("KMP_TASK_MAXIMUM_READY_PER_THREAD");
+  int max_tasks_ready =
+      max_tasks_ready_str ? atoi(max_tasks_ready_str) : MAX_TASKS_READY_DEFAULT;
+  if (max_tasks_ready <= 0)
+    max_tasks_ready = 1;
+
+  /* check if throttling is enabled (it is by default) */
+  char *throttling_str = getenv("KMP_ENABLE_TASK_THROTTLING");
+  int throttling = throttling_str ? *throttling_str == '1' : 1;
+
+  volatile int done = 0;
+
+/* testing KMP_TASK_MAXIMUM_READY */
+#pragma omp parallel num_threads(2) default(none)                              \
+    shared(max_tasks_ready, throttling, done)
+  {
+    if (omp_get_thread_num() == 1)
+      while (!done)
+        ;
+
+#pragma omp master
+    {
+      int ntasks = 0;
+      while (!done) {
+#pragma omp task default(none) shared(done)
+        done = 1;
+
+        assert(++ntasks <= max_tasks_ready + 1);
+      }
+    }
+  }
+
+  return 0;
+}