OpenMP Wait/release improvements.

[jlpeyton]

Summary

These changes improve the wait/release mechanism for threads spinning in barriers that are handling tasks while spinnin by providing feedback to the barriers about any task stealing that occurs.

[AndreyChurbanov]

LGTM

openmp/trunk/runtime/src/kmp_barrier.cpp

 //
 //===----------------------------------------------------------------------===//
 
 
 #include "kmp.h"
 #include "kmp_wait_release.h"
 #include "kmp_stats.h"
 #include "kmp_itt.h"
+#include "kmp_os.h"
+
 
 #if KMP_MIC
 #include <immintrin.h>
 #define USE_NGO_STORES 1
 #endif // KMP_MIC
 
 #if KMP_MIC && USE_NGO_STORES
 // ICV copying
else { // Worker threads
             TCW_8(thr_bar->b_go, KMP_INIT_BARRIER_STATE); // Reset my b_go flag for next time
         }
         else { // Thread barrier data is initialized, this is a leaf, blocktime is infinite, not nested
             // Wait on my "offset" bits on parent's b_go flag
             thr_bar->wait_flag = KMP_BARRIER_PARENT_FLAG;
             kmp_flag_oncore flag(&thr_bar->parent_bar->b_go, KMP_BARRIER_STATE_BUMP, thr_bar->offset,
                                  bt, this_thr
                                  USE_ITT_BUILD_ARG(itt_sync_obj) );
-            flag.wait(this_thr, TRUE
-                      USE_ITT_BUILD_ARG(itt_sync_obj) );
+            flag.wait(this_thr, TRUE);
             if (thr_bar->wait_flag == KMP_BARRIER_SWITCHING) { // Thread was switched to own b_go
                 TCW_8(thr_bar->b_go, KMP_INIT_BARRIER_STATE); // Reset my b_go flag for next time
             }
             else { // Reset my bits on parent's b_go flag
                 ((char*)&(thr_bar->parent_bar->b_go))[thr_bar->offset] = 0;
             }
         }
         thr_bar->wait_flag = KMP_BARRIER_NOT_WAITING;
         // Early exit for reaping threads releasing forkjoin barrier
         if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
             return;
         // The worker thread may now assume that the team is valid.
         team = __kmp_threads[gtid]->th.th_team;
         KMP_DEBUG_ASSERT(team != NULL);
         tid = __kmp_tid_from_gtid(gtid);
 
         KA_TRACE(20, ("__kmp_hierarchical_barrier_release: T#%d(%d:%d) set go(%p) = %u\n",
                       gtid, team->t.t_id, tid, &thr_bar->b_go, KMP_INIT_BARRIER_STATE));
         KMP_MB();  // Flush all pending memory write invalidates.
     }
 
+    nproc = this_thr->th.th_team_nproc;
     int level = team->t.t_level;
 #if OMP_40_ENABLED
     if (team->t.t_threads[0]->th.th_teams_microtask ) {    // are we inside the teams construct?
         if (team->t.t_pkfn != (microtask_t)__kmp_teams_master && this_thr->th.th_teams_level == level)
             ++level; // level was not increased in teams construct for team_of_workers
         if( this_thr->th.th_teams_size.nteams > 1 )
             ++level; // level was not increased in teams construct for team_of_masters
     }
 #endif
     if (level == 1) thr_bar->use_oncore_barrier = 1;
     else thr_bar->use_oncore_barrier = 0; // Do not use oncore barrier when nested
-    nproc = this_thr->th.th_team_nproc;
 
     // If the team size has increased, we still communicate with old leaves via oncore barrier.
     unsigned short int old_leaf_kids = thr_bar->leaf_kids;
     kmp_uint64 old_leaf_state = thr_bar->leaf_state;
     team_change = __kmp_init_hierarchical_barrier_thread(bt, thr_bar, nproc, gtid, tid, team);
     // But if the entire team changes, we won't use oncore barrier at all
     if (team_change) old_leaf_kids = 0;
 

openmp/trunk/runtime/src/kmp_os.h

 #  define KMP_COMPARE_AND_STORE_PTR(p, cv, sv)  __kmp_compare_and_store64( (volatile kmp_int64*)(p), (kmp_int64)(cv), (kmp_int64)(sv) )
 # endif /* KMP_ARCH_X86 */
 
 # define KMP_COMPARE_AND_STORE_RET8(p, cv, sv)  __kmp_compare_and_store_ret8( (p), (cv), (sv) )
 # define KMP_COMPARE_AND_STORE_RET16(p, cv, sv) __kmp_compare_and_store_ret16( (p), (cv), (sv) )
 //# define KMP_COMPARE_AND_STORE_RET32(p, cv, sv) __kmp_compare_and_store_ret32( (p), (cv), (sv) )
 # define KMP_COMPARE_AND_STORE_RET64(p, cv, sv) __kmp_compare_and_store_ret64( (p), (cv), (sv) )
 
-# define KMP_XCHG_FIXED8(p, v)                  __kmp_xchg_fixed8( (p), (v) );
+# define KMP_XCHG_FIXED8(p, v)                  __kmp_xchg_fixed8( (volatile kmp_int8*)(p), (kmp_int8)(v) );
 # define KMP_XCHG_FIXED16(p, v)                 __kmp_xchg_fixed16( (p), (v) );
 //# define KMP_XCHG_FIXED32(p, v)                 __kmp_xchg_fixed32( (p), (v) );
 //# define KMP_XCHG_FIXED64(p, v)                 __kmp_xchg_fixed64( (p), (v) );
 //# define KMP_XCHG_REAL32(p, v)                  __kmp_xchg_real32( (p), (v) );
 # define KMP_XCHG_REAL64(p, v)                  __kmp_xchg_real64( (p), (v) );
 
 
 #elif (KMP_ASM_INTRINS && KMP_OS_UNIX) || !(KMP_ARCH_X86 || KMP_ARCH_X86_64)

 #  define KMP_COMPARE_AND_STORE_PTR(p, cv, sv)  __kmp_compare_and_store64( (volatile kmp_int64*)(p), (kmp_int64)(cv), (kmp_int64)(sv) )
 # endif /* KMP_ARCH_X86 */
 
 # define KMP_COMPARE_AND_STORE_RET8(p, cv, sv)  __kmp_compare_and_store_ret8( (p), (cv), (sv) )
 # define KMP_COMPARE_AND_STORE_RET16(p, cv, sv) __kmp_compare_and_store_ret16( (p), (cv), (sv) )
 # define KMP_COMPARE_AND_STORE_RET32(p, cv, sv) __kmp_compare_and_store_ret32( (p), (cv), (sv) )
 # define KMP_COMPARE_AND_STORE_RET64(p, cv, sv) __kmp_compare_and_store_ret64( (p), (cv), (sv) )
 
-# define KMP_XCHG_FIXED8(p, v)                  __kmp_xchg_fixed8( (p), (v) );
+# define KMP_XCHG_FIXED8(p, v)                  __kmp_xchg_fixed8( (volatile kmp_int8*)(p), (kmp_int8)(v) );
 # define KMP_XCHG_FIXED16(p, v)                 __kmp_xchg_fixed16( (p), (v) );
 # define KMP_XCHG_FIXED32(p, v)                 __kmp_xchg_fixed32( (p), (v) );
 # define KMP_XCHG_FIXED64(p, v)                 __kmp_xchg_fixed64( (p), (v) );
 # define KMP_XCHG_REAL32(p, v)                  __kmp_xchg_real32( (p), (v) );
 # define KMP_XCHG_REAL64(p, v)                  __kmp_xchg_real64( (p), (v) );
 
 #endif /* KMP_ASM_INTRINS */
 

openmp/trunk/runtime/src/kmp_runtime.c

 #endif
 #if KMP_NESTED_HOT_TEAMS
             } // end of check of t_nproc vs. new_nproc vs. hot_team_nth
 #endif // KMP_NESTED_HOT_TEAMS
             /* make sure everyone is syncronized */
             __kmp_initialize_team( team, new_nproc, new_icvs, root->r.r_uber_thread->th.th_ident );
 
             if ( __kmp_tasking_mode != tskm_immediate_exec ) {
+                // Signal the worker threads to stop looking for tasks while spin waiting.
+                // The task teams are reference counted and will be deallocated by the last worker thread.
                 int tt_idx;
                 for (tt_idx=0; tt_idx<2; ++tt_idx) {
+                    // We don't know which of the two task teams workers are waiting on, so deactivate both.
                     kmp_task_team_t *task_team = team->t.t_task_team[tt_idx];
-                    if ( task_team != NULL ) {
-                        KMP_DEBUG_ASSERT( ! TCR_4(task_team->tt.tt_found_tasks) );
-                        task_team->tt.tt_nproc = new_nproc;
-                        task_team->tt.tt_unfinished_threads = new_nproc;
-                        task_team->tt.tt_ref_ct = new_nproc - 1;
+                    if ( (task_team != NULL) && TCR_SYNC_4(task_team->tt.tt_active) ) {
+                        TCW_SYNC_4( task_team->tt.tt_active, FALSE );
+                        team->t.t_task_team[tt_idx] = NULL;
                     }
                 }
             }
 
             /* reinitialize the old threads */
             if (level) {
                 for( f = 0  ;  f < team->t.t_nproc  ;  f++ ) {
                     __kmp_initialize_info( team->t.t_threads[ f ], team, f,

openmp/trunk/runtime/src/kmp_tasking.c

 static void __kmp_alloc_task_deque( kmp_info_t *thread, kmp_thread_data_t *thread_data );
 static int  __kmp_realloc_task_threads_data( kmp_info_t *thread, kmp_task_team_t *task_team );
 
 #ifdef OMP_41_ENABLED
 static void __kmp_bottom_half_finish_proxy( kmp_int32 gtid, kmp_task_t * ptask );
 #endif
 
 static inline void __kmp_null_resume_wrapper(int gtid, volatile void *flag) {
+    if (!flag) return;
     switch (((kmp_flag_64 *)flag)->get_type()) {
     case flag32: __kmp_resume_32(gtid, NULL); break;
     case flag64: __kmp_resume_64(gtid, NULL); break;
     case flag_oncore: __kmp_resume_oncore(gtid, NULL); break;
     }
 }
 
 #ifdef BUILD_TIED_TASK_STACK
if ( ( team->t.t_task_team[this_thr->th.th_task_state] == NULL ) && ( always || team->t.t_nproc > 1 ) ) {
         // the signal to stop checking for tasks (they can't safely
         // reference the kmp_team_t struct, which could be reallocated
         // by the master thread).
         team->t.t_task_team[this_thr->th.th_task_state] = __kmp_allocate_task_team( this_thr, team );
         KA_TRACE(20, ("__kmp_task_team_setup: Master T#%d created new task_team %p for team %d\n",
                       __kmp_gtid_from_thread(this_thr), team->t.t_task_team[this_thr->th.th_task_state],
                         ((team != NULL) ? team->t.t_id : -1)) );
     }
-    //else
-        // All threads have reported in, and no tasks were spawned
-        // for this release->gather region.  Leave the old task
-        // team struct in place for the upcoming region.  No task
-        // teams are formed for serialized teams.
+    // else: Either all threads have reported in, and no tasks were spawned for this release->gather region
+    // Leave the old task team struct in place for the upcoming region.
+    // No task teams are formed for serialized teams.
     if (both) {
         int other_team = 1 - this_thr->th.th_task_state;
         if ( ( team->t.t_task_team[other_team] == NULL ) && ( team->t.t_nproc > 1 ) ) { // setup other team as well
             team->t.t_task_team[other_team] = __kmp_allocate_task_team( this_thr, team );
             KA_TRACE( 20, ( "__kmp_task_team_setup: Master T#%d created new task_team %p for team %d\n",
                             __kmp_gtid_from_thread( this_thr ), team->t.t_task_team[other_team],
                             ((team != NULL) ? team->t.t_id : -1)) );
     }

openmp/trunk/runtime/src/kmp_wait_release.h

  public:
     typedef P flag_t;
     kmp_flag(volatile P *p, flag_type ft) : loc(p), t(ft) {}
     /*!
      * @result the pointer to the actual flag
      */
     volatile P * get() { return loc; }
     /*!
+     * @param new_loc in   set loc to point at new_loc
+     */
+    void set(volatile P *new_loc) { loc = new_loc; }
+    /*!
      * @result the flag_type
      */
     flag_type get_type() { return t; }
     // Derived classes must provide the following:
     /*
     kmp_info_t * get_waiter(kmp_uint32 i);
     kmp_uint32 get_num_waiters();
     bool done_check();
     bool done_check_val(P old_loc);
     bool notdone_check();
     P internal_release();
+    void suspend(int th_gtid);
+    void resume(int th_gtid);
     P set_sleeping();
     P unset_sleeping();
     bool is_sleeping();
+    bool is_any_sleeping();
     bool is_sleeping_val(P old_loc);
+    int execute_tasks(kmp_info_t *this_thr, kmp_int32 gtid, int final_spin, int *thread_finished
+                      USE_ITT_BUILD_ARG(void * itt_sync_obj), kmp_int32 is_constrained);
     */
 };
 
 /* Spin wait loop that first does pause, then yield, then sleep. A thread that calls __kmp_wait_*
    must make certain that another thread calls __kmp_release to wake it back up to prevent deadlocks!  */
 template <class C>
 static inline void __kmp_wait_template(kmp_info_t *this_thr, C *flag, int final_spin
                                        USE_ITT_BUILD_ARG(void * itt_sync_obj) )
#ifdef KMP_DEBUG
     // FIX ME
     kmp_info_t * wait_thr = flag->get_waiter(0);
     int target_gtid = wait_thr->th.th_info.ds.ds_gtid;
     int gtid = TCR_4(__kmp_init_gtid) ? __kmp_get_gtid() : -1;
 #endif
     KF_TRACE(20, ("__kmp_release: T#%d releasing T#%d spin(%p)\n", gtid, target_gtid, flag->get()));
     KMP_DEBUG_ASSERT(flag->get());
     KMP_FSYNC_RELEASING(flag->get());
     
-    typename C::flag_t old_spin = flag->internal_release();
+    flag->internal_release();
     
-    KF_TRACE(100, ("__kmp_release: T#%d old spin(%p)=%d, set new spin=%d\n",
-                   gtid, flag->get(), old_spin, *(flag->get())));
+    KF_TRACE(100, ("__kmp_release: T#%d set new spin=%d\n", gtid, flag->get(), *(flag->get())));
     
     if (__kmp_dflt_blocktime != KMP_MAX_BLOCKTIME) {
         // Only need to check sleep stuff if infinite block time not set
-        if (flag->is_sleeping_val(old_spin)) {
+        if (flag->is_any_sleeping()) { // Are *any* of the threads that wait on this flag sleeping?
             for (unsigned int i=0; i<flag->get_num_waiters(); ++i) {
-                kmp_info_t * waiter = flag->get_waiter(i);
+                kmp_info_t * waiter = flag->get_waiter(i); // if a sleeping waiter exists at i, sets current_waiter to i inside the flag
+                if (waiter) {
                     int wait_gtid = waiter->th.th_info.ds.ds_gtid;
                     // Wake up thread if needed
-                KF_TRACE(50, ("__kmp_release: T#%d waking up thread T#%d since sleep spin(%p) set\n",
+                    KF_TRACE(50, ("__kmp_release: T#%d waking up thread T#%d since sleep flag(%p) set\n",
                                   gtid, wait_gtid, flag->get()));
-                flag->resume(wait_gtid);
+                    flag->resume(wait_gtid); // unsets flag's current_waiter when done
+                }
             }
-        } else {
-            KF_TRACE(50, ("__kmp_release: T#%d don't wake up thread T#%d since sleep spin(%p) not set\n",
-                          gtid, target_gtid, flag->get()));
         }
     }
 }
 
 template <typename FlagType>
 struct flag_traits {};
 
 template <>
public:
      * while (flag.notdone_check()) { pause(); }
      * @endcode
      */
     bool notdone_check() { return traits_type::tcr(*(this->get())) != checker; }
     /*!
      * @result Actual flag value before release was applied.
      * Trigger all waiting threads to run by modifying flag to release state.
      */
-    FlagType internal_release() {
-        return traits_type::test_then_add4((volatile FlagType *)this->get());
+    void internal_release() {
+        (void) traits_type::test_then_add4((volatile FlagType *)this->get());
     }
     /*!
      * @result Actual flag value before sleep bit(s) set.
      * Notes that there is at least one thread sleeping on the flag by setting sleep bit(s).
      */
     FlagType set_sleeping() { 
         return traits_type::test_then_or((volatile FlagType *)this->get(), KMP_BARRIER_SLEEP_STATE);
     }
     /*!
      * @result Actual flag value before sleep bit(s) cleared.
      * Notes that there are no longer threads sleeping on the flag by clearing sleep bit(s).
      */
     FlagType unset_sleeping() { 
         return traits_type::test_then_and((volatile FlagType *)this->get(), ~KMP_BARRIER_SLEEP_STATE);
     }
     /*! 
      * @param old_loc in   old value of flag
      * Test whether there are threads sleeping on the flag's old value in old_loc.
      */
     bool is_sleeping_val(FlagType old_loc) { return old_loc & KMP_BARRIER_SLEEP_STATE; }
     /*! 
      * Test whether there are threads sleeping on the flag.
      */
     bool is_sleeping() { return is_sleeping_val(*(this->get())); }
+    bool is_any_sleeping() { return is_sleeping_val(*(this->get())); }
+    kmp_uint8 *get_stolen() { return NULL; }
+    enum barrier_type get_bt() { return bs_last_barrier; }
 };
 
 class kmp_flag_32 : public kmp_basic_flag<kmp_uint32> {
 public:
     kmp_flag_32(volatile kmp_uint32 *p) : kmp_basic_flag<kmp_uint32>(p) {}
     kmp_flag_32(volatile kmp_uint32 *p, kmp_info_t *thr) : kmp_basic_flag<kmp_uint32>(p, thr) {}
     kmp_flag_32(volatile kmp_uint32 *p, kmp_uint32 c) : kmp_basic_flag<kmp_uint32>(p, c) {}
     void suspend(int th_gtid) { __kmp_suspend_32(th_gtid, this); }
bool notdone_check() {
             __kmp_wait_64(this_thr, &flag, TRUE
 #if USE_ITT_BUILD
                           , itt_sync_obj
 #endif
                           );
         }
         return false;
     }
-    kmp_uint64 internal_release() { 
-        kmp_uint64 old_val;
+    void internal_release() {
         if (__kmp_dflt_blocktime == KMP_MAX_BLOCKTIME) {
-            old_val = *get();
             byteref(get(),offset) = 1;
         }
         else {
             kmp_uint64 mask=0;
             byteref(&mask,offset) = 1;
-            old_val = KMP_TEST_THEN_OR64((volatile kmp_int64 *)get(), mask);
+            (void) KMP_TEST_THEN_OR64((volatile kmp_int64 *)get(), mask);
         }
-        return old_val;
     }
     kmp_uint64 set_sleeping() { 
         return KMP_TEST_THEN_OR64((kmp_int64 volatile *)get(), KMP_BARRIER_SLEEP_STATE);
     }
     kmp_uint64 unset_sleeping() { 
         return KMP_TEST_THEN_AND64((kmp_int64 volatile *)get(), ~KMP_BARRIER_SLEEP_STATE);
     }
     bool is_sleeping_val(kmp_uint64 old_loc) { return old_loc & KMP_BARRIER_SLEEP_STATE; }
     bool is_sleeping() { return is_sleeping_val(*get()); }
-    void wait(kmp_info_t *this_thr, int final_spin
-              USE_ITT_BUILD_ARG(void * itt_sync_obj)) {
-        __kmp_wait_template(this_thr, this, final_spin
+    bool is_any_sleeping() { return is_sleeping_val(*get()); }
+    void wait(kmp_info_t *this_thr, int final_spin) {
+        __kmp_wait_template<kmp_flag_oncore>(this_thr, this, final_spin
                             USE_ITT_BUILD_ARG(itt_sync_obj));
     }
     void release() { __kmp_release_template(this); }
     void suspend(int th_gtid) { __kmp_suspend_oncore(th_gtid, this); }
     void resume(int th_gtid) { __kmp_resume_oncore(th_gtid, this); }
     int execute_tasks(kmp_info_t *this_thr, kmp_int32 gtid, int final_spin, int *thread_finished
                       USE_ITT_BUILD_ARG(void * itt_sync_obj), kmp_int32 is_constrained) {
         return __kmp_execute_tasks_oncore(this_thr, gtid, this, final_spin, thread_finished
                                       USE_ITT_BUILD_ARG(itt_sync_obj), is_constrained);
     }
+    kmp_uint8 *get_stolen() { return NULL; }
+    enum barrier_type get_bt() { return bt; }
 };
 
 /*!
 @}
 */
 
 #endif // KMP_WAIT_RELEASE_H