kmp_barrier.cpp

/*
 * kmp_barrier.cpp
 */
 
//===----------------------------------------------------------------------===//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
 
#include "kmp.h"
#include "kmp_wait_release.h"
#include "kmp_itt.h"
#include "kmp_os.h"
#include "kmp_stats.h"
#include "ompt-specific.h"
 
#if KMP_MIC
#include <immintrin.h>
#define USE_NGO_STORES 1
#endif // KMP_MIC
 
#if KMP_MIC && USE_NGO_STORES
// ICV copying
#define ngo_load(src) __m512d Vt = _mm512_load_pd((void *)(src))
#define ngo_store_icvs(dst, src) _mm512_storenrngo_pd((void *)(dst), Vt)
#define ngo_store_go(dst, src) _mm512_storenrngo_pd((void *)(dst), Vt)
#define ngo_sync() __asm__ volatile("lock; addl $0,0(%%rsp)" ::: "memory")
#else
#define ngo_load(src) ((void)0)
#define ngo_store_icvs(dst, src) copy_icvs((dst), (src))
#define ngo_store_go(dst, src) KMP_MEMCPY((dst), (src), CACHE_LINE)
#define ngo_sync() ((void)0)
#endif /* KMP_MIC && USE_NGO_STORES */
 
void __kmp_print_structure(void); // Forward declaration
 
// ---------------------------- Barrier Algorithms ----------------------------
 
// Linear Barrier
template <bool cancellable = false>
static bool __kmp_linear_barrier_gather_template(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    void (*reduce)(void *, void *) USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  KMP_TIME_DEVELOPER_PARTITIONED_BLOCK(KMP_linear_gather);
  kmp_team_t *team = this_thr->th.th_team;
  kmp_bstate_t *thr_bar = &this_thr->th.th_bar[bt].bb;
  kmp_info_t **other_threads = team->t.t_threads;
 
  KA_TRACE(
      20,
      ("__kmp_linear_barrier_gather: T#%d(%d:%d) enter for barrier type %d\n",
       gtid, team->t.t_id, tid, bt));
  KMP_DEBUG_ASSERT(this_thr == other_threads[this_thr->th.th_info.ds.ds_tid]);
 
#if USE_ITT_BUILD && USE_ITT_NOTIFY
  // Barrier imbalance - save arrive time to the thread
  if (__kmp_forkjoin_frames_mode == 3 || __kmp_forkjoin_frames_mode == 2) {
    this_thr->th.th_bar_arrive_time = this_thr->th.th_bar_min_time =
        __itt_get_timestamp();
  }
#endif
  // We now perform a linear reduction to signal that all of the threads have
  // arrived.
  if (!KMP_MASTER_TID(tid)) {
    KA_TRACE(20,
             ("__kmp_linear_barrier_gather: T#%d(%d:%d) releasing T#%d(%d:%d)"
              "arrived(%p): %llu => %llu\n",
              gtid, team->t.t_id, tid, __kmp_gtid_from_tid(0, team),
              team->t.t_id, 0, &thr_bar->b_arrived, thr_bar->b_arrived,
              thr_bar->b_arrived + KMP_BARRIER_STATE_BUMP));
    // Mark arrival to primary thread
    /* After performing this write, a worker thread may not assume that the team
       is valid any more - it could be deallocated by the primary thread at any
       time. */
    kmp_flag_64<> flag(&thr_bar->b_arrived, other_threads[0]);
    flag.release();
  } else {
    kmp_balign_team_t *team_bar = &team->t.t_bar[bt];
    int nproc = this_thr->th.th_team_nproc;
    int i;
    // Don't have to worry about sleep bit here or atomic since team setting
    kmp_uint64 new_state = team_bar->b_arrived + KMP_BARRIER_STATE_BUMP;
 
    // Collect all the worker team member threads.
    for (i = 1; i < nproc; ++i) {
#if KMP_CACHE_MANAGE
      // Prefetch next thread's arrived count
      if (i + 1 < nproc)
        KMP_CACHE_PREFETCH(&other_threads[i + 1]->th.th_bar[bt].bb.b_arrived);
#endif /* KMP_CACHE_MANAGE */
      KA_TRACE(20, ("__kmp_linear_barrier_gather: T#%d(%d:%d) wait T#%d(%d:%d) "
                    "arrived(%p) == %llu\n",
                    gtid, team->t.t_id, tid, __kmp_gtid_from_tid(i, team),
                    team->t.t_id, i,
                    &other_threads[i]->th.th_bar[bt].bb.b_arrived, new_state));
 
      // Wait for worker thread to arrive
      if (cancellable) {
        kmp_flag_64<true, false> flag(
            &other_threads[i]->th.th_bar[bt].bb.b_arrived, new_state);
        if (flag.wait(this_thr, FALSE USE_ITT_BUILD_ARG(itt_sync_obj)))
          return true;
      } else {
        kmp_flag_64<> flag(&other_threads[i]->th.th_bar[bt].bb.b_arrived,
                           new_state);
        flag.wait(this_thr, FALSE USE_ITT_BUILD_ARG(itt_sync_obj));
      }
#if USE_ITT_BUILD && USE_ITT_NOTIFY
      // Barrier imbalance - write min of the thread time and the other thread
      // time to the thread.
      if (__kmp_forkjoin_frames_mode == 2) {
        this_thr->th.th_bar_min_time = KMP_MIN(
            this_thr->th.th_bar_min_time, other_threads[i]->th.th_bar_min_time);
      }
#endif
      if (reduce) {
        KA_TRACE(100,
                 ("__kmp_linear_barrier_gather: T#%d(%d:%d) += T#%d(%d:%d)\n",
                  gtid, team->t.t_id, tid, __kmp_gtid_from_tid(i, team),
                  team->t.t_id, i));
        OMPT_REDUCTION_DECL(this_thr, gtid);
        OMPT_REDUCTION_BEGIN;
        (*reduce)(this_thr->th.th_local.reduce_data,
                  other_threads[i]->th.th_local.reduce_data);
        OMPT_REDUCTION_END;
      }
    }
    // Don't have to worry about sleep bit here or atomic since team setting
    team_bar->b_arrived = new_state;
    KA_TRACE(20, ("__kmp_linear_barrier_gather: T#%d(%d:%d) set team %d "
                  "arrived(%p) = %llu\n",
                  gtid, team->t.t_id, tid, team->t.t_id, &team_bar->b_arrived,
                  new_state));
  }
  KA_TRACE(
      20,
      ("__kmp_linear_barrier_gather: T#%d(%d:%d) exit for barrier type %d\n",
       gtid, team->t.t_id, tid, bt));
  return false;
}
 
template <bool cancellable = false>
static bool __kmp_linear_barrier_release_template(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    int propagate_icvs USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  KMP_TIME_DEVELOPER_PARTITIONED_BLOCK(KMP_linear_release);
  kmp_bstate_t *thr_bar = &this_thr->th.th_bar[bt].bb;
  kmp_team_t *team;
 
  if (KMP_MASTER_TID(tid)) {
    unsigned int i;
    kmp_uint32 nproc = this_thr->th.th_team_nproc;
    kmp_info_t **other_threads;
 
    team = __kmp_threads[gtid]->th.th_team;
    KMP_DEBUG_ASSERT(team != NULL);
    other_threads = team->t.t_threads;
 
    KA_TRACE(20, ("__kmp_linear_barrier_release: T#%d(%d:%d) primary enter for "
                  "barrier type %d\n",
                  gtid, team->t.t_id, tid, bt));
 
    if (nproc > 1) {
#if KMP_BARRIER_ICV_PUSH
      {
        KMP_TIME_DEVELOPER_PARTITIONED_BLOCK(USER_icv_copy);
        if (propagate_icvs) {
          ngo_load(&team->t.t_implicit_task_taskdata[0].td_icvs);
          for (i = 1; i < nproc; ++i) {
            __kmp_init_implicit_task(team->t.t_ident, team->t.t_threads[i],
                                     team, i, FALSE);
            ngo_store_icvs(&team->t.t_implicit_task_taskdata[i].td_icvs,
                           &team->t.t_implicit_task_taskdata[0].td_icvs);
          }
          ngo_sync();
        }
      }
#endif // KMP_BARRIER_ICV_PUSH
 
      // Now, release all of the worker threads
      for (i = 1; i < nproc; ++i) {
#if KMP_CACHE_MANAGE
        // Prefetch next thread's go flag
        if (i + 1 < nproc)
          KMP_CACHE_PREFETCH(&other_threads[i + 1]->th.th_bar[bt].bb.b_go);
#endif /* KMP_CACHE_MANAGE */
        KA_TRACE(
            20,
            ("__kmp_linear_barrier_release: T#%d(%d:%d) releasing T#%d(%d:%d) "
             "go(%p): %u => %u\n",
             gtid, team->t.t_id, tid, other_threads[i]->th.th_info.ds.ds_gtid,
             team->t.t_id, i, &other_threads[i]->th.th_bar[bt].bb.b_go,
             other_threads[i]->th.th_bar[bt].bb.b_go,
             other_threads[i]->th.th_bar[bt].bb.b_go + KMP_BARRIER_STATE_BUMP));
        kmp_flag_64<> flag(&other_threads[i]->th.th_bar[bt].bb.b_go,
                           other_threads[i]);
        flag.release();
      }
    }
  } else { // Wait for the PRIMARY thread to release us
    KA_TRACE(20, ("__kmp_linear_barrier_release: T#%d wait go(%p) == %u\n",
                  gtid, &thr_bar->b_go, KMP_BARRIER_STATE_BUMP));
    if (cancellable) {
      kmp_flag_64<true, false> flag(&thr_bar->b_go, KMP_BARRIER_STATE_BUMP);
      if (flag.wait(this_thr, TRUE USE_ITT_BUILD_ARG(itt_sync_obj)))
        return true;
    } else {
      kmp_flag_64<> flag(&thr_bar->b_go, KMP_BARRIER_STATE_BUMP);
      flag.wait(this_thr, TRUE USE_ITT_BUILD_ARG(itt_sync_obj));
    }
#if USE_ITT_BUILD && USE_ITT_NOTIFY
    if ((__itt_sync_create_ptr && itt_sync_obj == NULL) || KMP_ITT_DEBUG) {
      // In a fork barrier; cannot get the object reliably (or ITTNOTIFY is
      // disabled)
      itt_sync_obj = __kmp_itt_barrier_object(gtid, bs_forkjoin_barrier, 0, -1);
      // Cancel wait on previous parallel region...
      __kmp_itt_task_starting(itt_sync_obj);
 
      if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
        return false;
 
      itt_sync_obj = __kmp_itt_barrier_object(gtid, bs_forkjoin_barrier);
      if (itt_sync_obj != NULL)
        // Call prepare as early as possible for "new" barrier
        __kmp_itt_task_finished(itt_sync_obj);
    } else
#endif /* USE_ITT_BUILD && USE_ITT_NOTIFY */
        // Early exit for reaping threads releasing forkjoin barrier
        if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
      return false;
// The worker thread may now assume that the team is valid.
#ifdef KMP_DEBUG
    tid = __kmp_tid_from_gtid(gtid);
    team = __kmp_threads[gtid]->th.th_team;
#endif
    KMP_DEBUG_ASSERT(team != NULL);
    TCW_4(thr_bar->b_go, KMP_INIT_BARRIER_STATE);
    KA_TRACE(20,
             ("__kmp_linear_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.
  }
  KA_TRACE(
      20,
      ("__kmp_linear_barrier_release: T#%d(%d:%d) exit for barrier type %d\n",
       gtid, team->t.t_id, tid, bt));
  return false;
}
 
static void __kmp_linear_barrier_gather(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    void (*reduce)(void *, void *) USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  __kmp_linear_barrier_gather_template<false>(
      bt, this_thr, gtid, tid, reduce USE_ITT_BUILD_ARG(itt_sync_obj));
}
 
static bool __kmp_linear_barrier_gather_cancellable(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    void (*reduce)(void *, void *) USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  return __kmp_linear_barrier_gather_template<true>(
      bt, this_thr, gtid, tid, reduce USE_ITT_BUILD_ARG(itt_sync_obj));
}
 
static void __kmp_linear_barrier_release(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    int propagate_icvs USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  __kmp_linear_barrier_release_template<false>(
      bt, this_thr, gtid, tid, propagate_icvs USE_ITT_BUILD_ARG(itt_sync_obj));
}
 
static bool __kmp_linear_barrier_release_cancellable(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    int propagate_icvs USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  return __kmp_linear_barrier_release_template<true>(
      bt, this_thr, gtid, tid, propagate_icvs USE_ITT_BUILD_ARG(itt_sync_obj));
}
 
// Tree barrier
static void __kmp_tree_barrier_gather(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    void (*reduce)(void *, void *) USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  KMP_TIME_DEVELOPER_PARTITIONED_BLOCK(KMP_tree_gather);
  kmp_team_t *team = this_thr->th.th_team;
  kmp_bstate_t *thr_bar = &this_thr->th.th_bar[bt].bb;
  kmp_info_t **other_threads = team->t.t_threads;
  kmp_uint32 nproc = this_thr->th.th_team_nproc;
  kmp_uint32 branch_bits = __kmp_barrier_gather_branch_bits[bt];
  kmp_uint32 branch_factor = 1 << branch_bits;
  kmp_uint32 child;
  kmp_uint32 child_tid;
  kmp_uint64 new_state = 0;
 
  KA_TRACE(
      20, ("__kmp_tree_barrier_gather: T#%d(%d:%d) enter for barrier type %d\n",
           gtid, team->t.t_id, tid, bt));
  KMP_DEBUG_ASSERT(this_thr == other_threads[this_thr->th.th_info.ds.ds_tid]);
 
#if USE_ITT_BUILD && USE_ITT_NOTIFY
  // Barrier imbalance - save arrive time to the thread
  if (__kmp_forkjoin_frames_mode == 3 || __kmp_forkjoin_frames_mode == 2) {
    this_thr->th.th_bar_arrive_time = this_thr->th.th_bar_min_time =
        __itt_get_timestamp();
  }
#endif
  // Perform tree gather to wait until all threads have arrived; reduce any
  // required data as we go
  child_tid = (tid << branch_bits) + 1;
  if (child_tid < nproc) {
    // Parent threads wait for all their children to arrive
    new_state = team->t.t_bar[bt].b_arrived + KMP_BARRIER_STATE_BUMP;
    child = 1;
    do {
      kmp_info_t *child_thr = other_threads[child_tid];
      kmp_bstate_t *child_bar = &child_thr->th.th_bar[bt].bb;
#if KMP_CACHE_MANAGE
      // Prefetch next thread's arrived count
      if (child + 1 <= branch_factor && child_tid + 1 < nproc)
        KMP_CACHE_PREFETCH(
            &other_threads[child_tid + 1]->th.th_bar[bt].bb.b_arrived);
#endif /* KMP_CACHE_MANAGE */
      KA_TRACE(20,
               ("__kmp_tree_barrier_gather: T#%d(%d:%d) wait T#%d(%d:%u) "
                "arrived(%p) == %llu\n",
                gtid, team->t.t_id, tid, __kmp_gtid_from_tid(child_tid, team),
                team->t.t_id, child_tid, &child_bar->b_arrived, new_state));
      // Wait for child to arrive
      kmp_flag_64<> flag(&child_bar->b_arrived, new_state);
      flag.wait(this_thr, FALSE USE_ITT_BUILD_ARG(itt_sync_obj));
#if USE_ITT_BUILD && USE_ITT_NOTIFY
      // Barrier imbalance - write min of the thread time and a child time to
      // the thread.
      if (__kmp_forkjoin_frames_mode == 2) {
        this_thr->th.th_bar_min_time = KMP_MIN(this_thr->th.th_bar_min_time,
                                               child_thr->th.th_bar_min_time);
      }
#endif
      if (reduce) {
        KA_TRACE(100,
                 ("__kmp_tree_barrier_gather: T#%d(%d:%d) += T#%d(%d:%u)\n",
                  gtid, team->t.t_id, tid, __kmp_gtid_from_tid(child_tid, team),
                  team->t.t_id, child_tid));
        OMPT_REDUCTION_DECL(this_thr, gtid);
        OMPT_REDUCTION_BEGIN;
        (*reduce)(this_thr->th.th_local.reduce_data,
                  child_thr->th.th_local.reduce_data);
        OMPT_REDUCTION_END;
      }
      child++;
      child_tid++;
    } while (child <= branch_factor && child_tid < nproc);
  }
 
  if (!KMP_MASTER_TID(tid)) { // Worker threads
    kmp_int32 parent_tid = (tid - 1) >> branch_bits;
 
    KA_TRACE(20,
             ("__kmp_tree_barrier_gather: T#%d(%d:%d) releasing T#%d(%d:%d) "
              "arrived(%p): %llu => %llu\n",
              gtid, team->t.t_id, tid, __kmp_gtid_from_tid(parent_tid, team),
              team->t.t_id, parent_tid, &thr_bar->b_arrived, thr_bar->b_arrived,
              thr_bar->b_arrived + KMP_BARRIER_STATE_BUMP));
 
    // Mark arrival to parent thread
    /* After performing this write, a worker thread may not assume that the team
       is valid any more - it could be deallocated by the primary thread at any
       time.  */
    kmp_flag_64<> flag(&thr_bar->b_arrived, other_threads[parent_tid]);
    flag.release();
  } else {
    // Need to update the team arrived pointer if we are the primary thread
    if (nproc > 1) // New value was already computed above
      team->t.t_bar[bt].b_arrived = new_state;
    else
      team->t.t_bar[bt].b_arrived += KMP_BARRIER_STATE_BUMP;
    KA_TRACE(20, ("__kmp_tree_barrier_gather: T#%d(%d:%d) set team %d "
                  "arrived(%p) = %llu\n",
                  gtid, team->t.t_id, tid, team->t.t_id,
                  &team->t.t_bar[bt].b_arrived, team->t.t_bar[bt].b_arrived));
  }
  KA_TRACE(20,
           ("__kmp_tree_barrier_gather: T#%d(%d:%d) exit for barrier type %d\n",
            gtid, team->t.t_id, tid, bt));
}
 
static void __kmp_tree_barrier_release(
    enum barrier_type bt, kmp_info_t *this_thr, int gtid, int tid,
    int propagate_icvs USE_ITT_BUILD_ARG(void *itt_sync_obj)) {
  KMP_TIME_DEVELOPER_PARTITIONED_BLOCK(KMP_tree_release);
  kmp_team_t *team;
  kmp_bstate_t *thr_bar = &this_thr->th.th_bar[bt].bb;
  kmp_uint32 nproc;
  kmp_uint32 branch_bits = __kmp_barrier_release_branch_bits[bt];
  kmp_uint32 branch_factor = 1 << branch_bits;
  kmp_uint32 child;
  kmp_uint32 child_tid;
 
  // Perform a tree release for all of the threads that have been gathered
  if (!KMP_MASTER_TID(
          tid)) { // Handle fork barrier workers who aren't part of a team yet
    KA_TRACE(20, ("__kmp_tree_barrier_release: T#%d wait go(%p) == %u\n", gtid,
                  &thr_bar->b_go, KMP_BARRIER_STATE_BUMP));
    // Wait for parent thread to release us
    kmp_flag_64<> flag(&thr_bar->b_go, KMP_BARRIER_STATE_BUMP);
    flag.wait(this_thr, TRUE USE_ITT_BUILD_ARG(itt_sync_obj));
#if USE_ITT_BUILD && USE_ITT_NOTIFY
    if ((__itt_sync_create_ptr && itt_sync_obj == NULL) || KMP_ITT_DEBUG) {
      // In fork barrier where we could not get the object reliably (or
      // ITTNOTIFY is disabled)
      itt_sync_obj = __kmp_itt_barrier_object(gtid, bs_forkjoin_barrier, 0, -1);
      // Cancel wait on previous parallel region...
      __kmp_itt_task_starting(itt_sync_obj);
 
      if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
        return;
 
      itt_sync_obj = __kmp_itt_barrier_object(gtid, bs_forkjoin_barrier);
      if (itt_sync_obj != NULL)
        // Call prepare as early as possible for "new" barrier
        __kmp_itt_task_finished(itt_sync_obj);
    } else
#endif /* USE_ITT_BUILD && USE_ITT_NOTIFY */
        // Early exit for reaping threads releasing forkjoin barrier
        if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
      return;