Clean up load balancing dynamic mode

[tlwilmar]

Summary

This patch cleans up the bookkeeping code for the load balancing dynamic mode.

When a thread is moved to or from the thread pool, the th_active_in_pool flag and the __kmp_thread_pool_active_nth global counter are both updated. This removes the need for the corrective code in the main wait loop. Another global counter, __kmp_thread_pool_nth, was removed completely, as it was only used for debugging, but was not under KMP_DEBUG.

[tlwilmar]

*ping*

[AndreyChurbanov]

LGTM

openmp/trunk/runtime/src/exports_so.txt

#endif /* USE_DEBUGGER */
 
         #
         # Internal functions exported for testing purposes.
         #
         __kmp_get_reduce_method;
         ___kmp_allocate;
         ___kmp_free;
         __kmp_thread_pool;
-        __kmp_thread_pool_nth;
 
   __kmp_reset_stats;
 
 #if USE_ITT_BUILD
         #
         # ITT support.
         #
         # The following entry points are added so that the backtraces from

openmp/trunk/runtime/src/kmp.h

 extern volatile kmp_info_t *__kmp_thread_pool;
 extern kmp_info_t *__kmp_thread_pool_insert_pt;
 
 // total num threads reachable from some root thread including all root threads
 extern volatile int __kmp_nth;
 /* total number of threads reachable from some root thread including all root
    threads, and those in the thread pool */
 extern volatile int __kmp_all_nth;
-extern int __kmp_thread_pool_nth;
 extern std::atomic<int> __kmp_thread_pool_active_nth;
 
 extern kmp_root_t **__kmp_root; /* root of thread hierarchy */
 /* end data protected by fork/join lock */
 /* ------------------------------------------------------------------------- */
 
 #define __kmp_get_gtid() __kmp_get_global_thread_id()
 #define __kmp_entry_gtid() __kmp_get_global_thread_id_reg()

 extern void __kmp_common_initialize(void);
 extern void __kmp_common_destroy(void);
 extern void __kmp_common_destroy_gtid(int gtid);
 
 #if KMP_OS_UNIX
 extern void __kmp_register_atfork(void);
 #endif
 extern void __kmp_suspend_initialize(void);
+extern void __kmp_suspend_initialize_thread(kmp_info_t *th);
 extern void __kmp_suspend_uninitialize_thread(kmp_info_t *th);
 
 extern kmp_info_t *__kmp_allocate_thread(kmp_root_t *root, kmp_team_t *team,
                                          int tid);
 #if OMP_40_ENABLED
 extern kmp_team_t *
 __kmp_allocate_team(kmp_root_t *root, int new_nproc, int max_nproc,
 #if OMPT_SUPPORT

openmp/trunk/runtime/src/kmp_global.cpp

 KMP_ALIGN_CACHE
 kmp_info_t **__kmp_threads = NULL;
 kmp_root_t **__kmp_root = NULL;
 
 /* data read/written to often by masters */
 KMP_ALIGN_CACHE
 volatile int __kmp_nth = 0;
 volatile int __kmp_all_nth = 0;
-int __kmp_thread_pool_nth = 0;
 volatile kmp_info_t *__kmp_thread_pool = NULL;
 volatile kmp_team_t *__kmp_team_pool = NULL;
 
 KMP_ALIGN_CACHE
 std::atomic<int> __kmp_thread_pool_active_nth = ATOMIC_VAR_INIT(0);
 
 /* -------------------------------------------------
  * GLOBAL/ROOT STATE */

openmp/trunk/runtime/src/kmp_runtime.cpp

#endif
   /* first, try to get one from the thread pool */
   if (__kmp_thread_pool) {
     new_thr = CCAST(kmp_info_t *, __kmp_thread_pool);
     __kmp_thread_pool = (volatile kmp_info_t *)new_thr->th.th_next_pool;
     if (new_thr == __kmp_thread_pool_insert_pt) {
       __kmp_thread_pool_insert_pt = NULL;
     }
     TCW_4(new_thr->th.th_in_pool, FALSE);
-    // Don't touch th_active_in_pool or th_active.
-    // The worker thread adjusts those flags as it sleeps/awakens.
-    __kmp_thread_pool_nth--;
+    __kmp_suspend_initialize_thread(new_thr);
+    __kmp_lock_suspend_mx(new_thr);
+    if (new_thr->th.th_active_in_pool == TRUE) {
+      KMP_DEBUG_ASSERT(new_thr->th.th_active == TRUE);
+      KMP_ATOMIC_DEC(&__kmp_thread_pool_active_nth);
+      new_thr->th.th_active_in_pool = FALSE;
+    }
+#if KMP_DEBUG
+    else {
+      KMP_DEBUG_ASSERT(new_thr->th.th_active == FALSE);
+    }
+#endif
+    __kmp_unlock_suspend_mx(new_thr);
 
     KA_TRACE(20, ("__kmp_allocate_thread: T#%d using thread T#%d\n",
                   __kmp_get_gtid(), new_thr->th.th_info.ds.ds_gtid));
     KMP_ASSERT(!new_thr->th.th_team);
     KMP_DEBUG_ASSERT(__kmp_nth < __kmp_threads_capacity);
-    KMP_DEBUG_ASSERT(__kmp_thread_pool_nth >= 0);
 
     /* setup the thread structure */
     __kmp_initialize_info(new_thr, team, new_tid,
                           new_thr->th.th_info.ds.ds_gtid);
     KMP_DEBUG_ASSERT(new_thr->th.th_serial_team);
 
     TCW_4(__kmp_nth, __kmp_nth + 1);
 
void __kmp_free_thread(kmp_info_t *this_th) {
   // Insert the new element on the list, and set __kmp_thread_pool_insert_pt
   // to its address.
   TCW_PTR(this_th->th.th_next_pool, *scan);
   __kmp_thread_pool_insert_pt = *scan = this_th;
   KMP_DEBUG_ASSERT((this_th->th.th_next_pool == NULL) ||
                    (this_th->th.th_info.ds.ds_gtid <
                     this_th->th.th_next_pool->th.th_info.ds.ds_gtid));
   TCW_4(this_th->th.th_in_pool, TRUE);
-  __kmp_thread_pool_nth++;
+  __kmp_suspend_initialize_thread(this_th);
+  __kmp_lock_suspend_mx(this_th);
+  if (this_th->th.th_active == TRUE) {
+    KMP_ATOMIC_INC(&__kmp_thread_pool_active_nth);
+    this_th->th.th_active_in_pool = TRUE;
+  }
+#if KMP_DEBUG
+  else {
+    KMP_DEBUG_ASSERT(this_th->th.th_active_in_pool == FALSE);
+  }
+#endif
+  __kmp_unlock_suspend_mx(this_th);
 
   TCW_4(__kmp_nth, __kmp_nth - 1);
 
 #ifdef KMP_ADJUST_BLOCKTIME
   /* Adjust blocktime back to user setting or default if necessary */
   /* Middle initialization might never have occurred                */
   if (!__kmp_env_blocktime && (__kmp_avail_proc > 0)) {
     KMP_DEBUG_ASSERT(__kmp_avail_proc > 0);
if (!is_root) {
     //
     // Currently, this can only happen as the library is unloaded,
     // so there are no harmful side effects.
     if (thread->th.th_active_in_pool) {
       thread->th.th_active_in_pool = FALSE;
       KMP_ATOMIC_DEC(&__kmp_thread_pool_active_nth);
       KMP_DEBUG_ASSERT(__kmp_thread_pool_active_nth >= 0);
     }
-
-    // Decrement # of [worker] threads in the pool.
-    KMP_DEBUG_ASSERT(__kmp_thread_pool_nth > 0);
-    --__kmp_thread_pool_nth;
   }
 
   __kmp_free_implicit_task(thread);
 
 // Free the fast memory for tasking
 #if USE_FAST_MEMORY
   __kmp_free_fast_memory(thread);
 #endif /* USE_FAST_MEMORY */
#endif
     while (__kmp_thread_pool != NULL) { // Loop thru all the thread in the pool.
       // Get the next thread from the pool.
       kmp_info_t *thread = CCAST(kmp_info_t *, __kmp_thread_pool);
       __kmp_thread_pool = thread->th.th_next_pool;
       // Reap it.
       KMP_DEBUG_ASSERT(thread->th.th_reap_state == KMP_SAFE_TO_REAP);
       thread->th.th_next_pool = NULL;
       thread->th.th_in_pool = FALSE;
+      thread->th.th_active_in_pool = FALSE;
+      KMP_ATOMIC_DEC(&__kmp_thread_pool_active_nth);
       __kmp_reap_thread(thread, 0);
     }
     __kmp_thread_pool_insert_pt = NULL;
 
     // Reap teams.
     while (__kmp_team_pool != NULL) { // Loop thru all the teams in the pool.
       // Get the next team from the pool.
       kmp_team_t *team = CCAST(kmp_team_t *, __kmp_team_pool);

openmp/trunk/runtime/src/kmp_wait_release.h

 #endif // KMP_USE_MONITOR
   }
 
   oversubscribed = (TCR_4(__kmp_nth) > __kmp_avail_proc);
   KMP_MB();
 
   // Main wait spin loop
   while (flag->notdone_check()) {
-    int in_pool;
     kmp_task_team_t *task_team = NULL;
     if (__kmp_tasking_mode != tskm_immediate_exec) {
       task_team = this_thr->th.th_task_team;
       /* If the thread's task team pointer is NULL, it means one of 3 things:
          1) A newly-created thread is first being released by
          __kmp_fork_barrier(), and its task team has not been set up yet.
          2) All tasks have been executed to completion.
          3) Tasking is off for this region.  This could be because we are in a
if (TCR_4(__kmp_global.g.g_done)) {
         __kmp_abort_thread();
       break;
     }
 
     // If we are oversubscribed, or have waited a bit (and
     // KMP_LIBRARY=throughput), then yield
     KMP_YIELD_OVERSUB_ELSE_SPIN(spins);
 
-    // Check if this thread was transferred from a team
-    // to the thread pool (or vice-versa) while spinning.
-    in_pool = !!TCR_4(this_thr->th.th_in_pool);
-    if (in_pool != !!this_thr->th.th_active_in_pool) {
-      if (in_pool) { // Recently transferred from team to pool
-        KMP_ATOMIC_INC(&__kmp_thread_pool_active_nth);
-        this_thr->th.th_active_in_pool = TRUE;
-        /* Here, we cannot assert that:
-           KMP_DEBUG_ASSERT(TCR_4(__kmp_thread_pool_active_nth) <=
-           __kmp_thread_pool_nth);
-           __kmp_thread_pool_nth is inc/dec'd by the master thread while the
-           fork/join lock is held, whereas __kmp_thread_pool_active_nth is
-           inc/dec'd asynchronously by the workers. The two can get out of sync
-           for brief periods of time.  */
-      } else { // Recently transferred from pool to team
-        KMP_ATOMIC_DEC(&__kmp_thread_pool_active_nth);
-        KMP_DEBUG_ASSERT(TCR_4(__kmp_thread_pool_active_nth) >= 0);
-        this_thr->th.th_active_in_pool = FALSE;
-      }
-    }
-
 #if KMP_STATS_ENABLED
     // Check if thread has been signalled to idle state
     // This indicates that the logical "join-barrier" has finished
     if (this_thr->th.th_stats->isIdle() &&
         KMP_GET_THREAD_STATE() == FORK_JOIN_BARRIER) {
       KMP_SET_THREAD_STATE(IDLE);
       KMP_PUSH_PARTITIONED_TIMER(OMP_idle);
     }

openmp/trunk/runtime/src/z_Linux_util.cpp

 void __kmp_suspend_initialize(void) {
   int status;
   status = pthread_mutexattr_init(&__kmp_suspend_mutex_attr);
   KMP_CHECK_SYSFAIL("pthread_mutexattr_init", status);
   status = pthread_condattr_init(&__kmp_suspend_cond_attr);
   KMP_CHECK_SYSFAIL("pthread_condattr_init", status);
 }
 
-static void __kmp_suspend_initialize_thread(kmp_info_t *th) {
+void __kmp_suspend_initialize_thread(kmp_info_t *th) {
   ANNOTATE_HAPPENS_AFTER(&th->th.th_suspend_init_count);
   if (th->th.th_suspend_init_count <= __kmp_fork_count) {
     /* this means we haven't initialized the suspension pthread objects for this
        thread in this instance of the process */
     int status;
     status = pthread_cond_init(&th->th.th_suspend_cv.c_cond,
                                &__kmp_suspend_cond_attr);
     KMP_CHECK_SYSFAIL("pthread_cond_init", status);

openmp/trunk/runtime/src/z_Windows_NT_util.cpp

void __kmp_disable(int *old_state) {
 
   if (__kmp_init_runtime)
     EnterCriticalSection(&__kmp_win32_section);
 }
 
 void __kmp_suspend_initialize(void) { /* do nothing */
 }
 
-static void __kmp_suspend_initialize_thread(kmp_info_t *th) {
+void __kmp_suspend_initialize_thread(kmp_info_t *th) {
   if (!TCR_4(th->th.th_suspend_init)) {
     /* this means we haven't initialized the suspension pthread objects for this
        thread in this instance of the process */
     __kmp_win32_cond_init(&th->th.th_suspend_cv);
     __kmp_win32_mutex_init(&th->th.th_suspend_mx);
     TCW_4(th->th.th_suspend_init, TRUE);
   }
 }