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_wait_release.h"
#include "kmp_barrier.h"
#include "kmp_itt.h"
#include "kmp_os.h"
#include "kmp_stats.h"
#include "ompt-specific.h"
// for distributed barrier
#include "kmp_affinity.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 ----------------------------
// Distributed barrier
 
// Compute how many threads to have polling each cache-line.
// We want to limit the number of writes to IDEAL_GO_RESOLUTION.
void distributedBarrier::computeVarsForN(size_t n) {
  int nsockets = 1;
  if (__kmp_topology) {
    int socket_level = __kmp_topology->get_level(KMP_HW_SOCKET);
    int core_level = __kmp_topology->get_level(KMP_HW_CORE);
    int ncores_per_socket =
        __kmp_topology->calculate_ratio(core_level, socket_level);
    nsockets = __kmp_topology->get_count(socket_level);
 
    if (nsockets <= 0)
      nsockets = 1;
    if (ncores_per_socket <= 0)
      ncores_per_socket = 1;
 
    threads_per_go = ncores_per_socket >> 1;
    if (!fix_threads_per_go) {
      // Minimize num_gos
      if (threads_per_go > 4) {
        if (KMP_OPTIMIZE_FOR_REDUCTIONS) {
          threads_per_go = threads_per_go >> 1;
        }
        if (threads_per_go > 4 && nsockets == 1)
          threads_per_go = threads_per_go >> 1;
      }
    }
    if (threads_per_go == 0)
      threads_per_go = 1;
    fix_threads_per_go = true;
    num_gos = n / threads_per_go;
    if (n % threads_per_go)
      num_gos++;
    if (nsockets == 1 || num_gos == 1)
      num_groups = 1;
    else {
      num_groups = num_gos / nsockets;
      if (num_gos % nsockets)
        num_groups++;
    }
    if (num_groups <= 0)
      num_groups = 1;
    gos_per_group = num_gos / num_groups;
    if (num_gos % num_groups)
      gos_per_group++;
    threads_per_group = threads_per_go * gos_per_group;
  } else {
    num_gos = n / threads_per_go;
    if (n % threads_per_go)
      num_gos++;
    if (num_gos == 1)
      num_groups = 1;
    else {
      num_groups = num_gos / 2;
      if (num_gos % 2)
        num_groups++;
    }
    gos_per_group = num_gos / num_groups;
    if (num_gos % num_groups)
      gos_per_group++;
    threads_per_group = threads_per_go * gos_per_group;
  }
}
 
void distributedBarrier::computeGo(size_t n) {
  // Minimize num_gos
  for (num_gos = 1;; num_gos++)
    if (IDEAL_CONTENTION * num_gos >= n)
      break;
  threads_per_go = n / num_gos;
  if (n % num_gos)
    threads_per_go++;
  while (num_gos > MAX_GOS) {
    threads_per_go++;
    num_gos = n / threads_per_go;
    if (n % threads_per_go)
      num_gos++;
  }
  computeVarsForN(n);
}
 
// This function is to resize the barrier arrays when the new number of threads
// exceeds max_threads, which is the current size of all the arrays
void distributedBarrier::resize(size_t nthr) {
  KMP_DEBUG_ASSERT(nthr > max_threads);
 
  // expand to requested size * 2
  max_threads = nthr * 2;
 
  // allocate arrays to new max threads
  for (int i = 0; i < MAX_ITERS; ++i) {
    if (flags[i])
      flags[i] = (flags_s *)KMP_INTERNAL_REALLOC(flags[i],
                                                 max_threads * sizeof(flags_s));
    else
      flags[i] = (flags_s *)KMP_INTERNAL_MALLOC(max_threads * sizeof(flags_s));
  }
 
  if (go)
    go = (go_s *)KMP_INTERNAL_REALLOC(go, max_threads * sizeof(go_s));
  else
    go = (go_s *)KMP_INTERNAL_MALLOC(max_threads * sizeof(go_s));
 
  if (iter)
    iter = (iter_s *)KMP_INTERNAL_REALLOC(iter, max_threads * sizeof(iter_s));
  else
    iter = (iter_s *)KMP_INTERNAL_MALLOC(max_threads * sizeof(iter_s));
 
  if (sleep)
    sleep =
        (sleep_s *)KMP_INTERNAL_REALLOC(sleep, max_threads * sizeof(sleep_s));
  else
    sleep = (sleep_s *)KMP_INTERNAL_MALLOC(max_threads * sizeof(sleep_s));
}
 
// This function is to set all the go flags that threads might be waiting
// on, and when blocktime is not infinite, it should be followed by a wake-up
// call to each thread
kmp_uint64 distributedBarrier::go_release() {
  kmp_uint64 next_go = iter[0].iter + distributedBarrier::MAX_ITERS;
  for (size_t j = 0; j < num_gos; j++) {
    go[j].go.store(next_go);
  }
  return next_go;
}
 
void distributedBarrier::go_reset() {
  for (size_t j = 0; j < max_threads; ++j) {
    for (size_t i = 0; i < distributedBarrier::MAX_ITERS; ++i) {
      flags[i][j].stillNeed = 1;
    }
    go[j].go.store(0);
    iter[j].iter = 0;
  }
}
 
// This function inits/re-inits the distributed barrier for a particular number
// of threads. If a resize of arrays is needed, it calls the resize function.
void distributedBarrier::init(size_t nthr) {
  size_t old_max = max_threads;
  if (nthr > max_threads) { // need more space in arrays
    resize(nthr);
  }
 
  for (size_t i = 0; i < max_threads; i++) {
    for (size_t j = 0; j < distributedBarrier::MAX_ITERS; j++) {
      flags[j][i].stillNeed = 1;
    }
    go[i].go.store(0);
    iter[i].iter = 0;
    if (i >= old_max)
      sleep[i].sleep = false;
  }
 
  // Recalculate num_gos, etc. based on new nthr
  computeVarsForN(nthr);
 
  num_threads = nthr;
 
  if (team_icvs == NULL)
    team_icvs = __kmp_allocate(sizeof(kmp_internal_control_t));
}
 
// This function is used only when KMP_BLOCKTIME is not infinite.
// static
void __kmp_dist_barrier_wakeup(enum barrier_type bt, kmp_team_t *team,
                               size_t start, size_t stop, size_t inc,
                               size_t tid) {
  KMP_DEBUG_ASSERT(__kmp_dflt_blocktime != KMP_MAX_BLOCKTIME);
  if (bt == bs_forkjoin_barrier && TCR_4(__kmp_global.g.g_done))
    return;
 
  kmp_info_t **other_threads = team->t.t_threads;
  for (size_t thr = start; thr < stop; thr += inc) {
    KMP_DEBUG_ASSERT(other_threads[thr]);
    int gtid = other_threads[thr]->th.th_info.ds.ds_gtid;
    // Wake up worker regardless of if it appears to be sleeping or not
    __kmp_atomic_resume_64(gtid, (kmp_atomic_flag_64<> *)NULL);
  }
}
 
static void __kmp_dist_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_dist_gather);
  kmp_team_t *team;
  distributedBarrier *b;
  kmp_info_t **other_threads;
  kmp_uint64 my_current_iter, my_next_iter;
  kmp_uint32 nproc;
  bool group_leader;
 
  team = this_thr->th.th_team;
  nproc = this_thr->th.th_team_nproc;
  other_threads = team->t.t_threads;
  b = team->t.b;
  my_current_iter = b->iter[tid].iter;
  my_next_iter = (my_current_iter + 1) % distributedBarrier::MAX_ITERS;
  group_leader = ((tid % b->threads_per_group) == 0);
 
  KA_TRACE(20,
           ("__kmp_dist_barrier_gather: T#%d(%d:%d) enter; barrier type %d\n",
            gtid, team->t.t_id, tid, bt));
 
#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
 
  if (group_leader) {
    // Start from the thread after the group leader
    size_t group_start = tid + 1;
    size_t group_end = tid + b->threads_per_group;
    size_t threads_pending = 0;
 
    if (group_end > nproc)
      group_end = nproc;
    do { // wait for threads in my group
      threads_pending = 0;
      // Check all the flags every time to avoid branch misspredict
      for (size_t thr = group_start; thr < group_end; thr++) {
        // Each thread uses a different cache line
        threads_pending += b->flags[my_current_iter][thr].stillNeed;
      }
      // Execute tasks here
      if (__kmp_tasking_mode != tskm_immediate_exec) {
        kmp_task_team_t *task_team = this_thr->th.th_task_team;
        if (task_team != NULL) {
          if (TCR_SYNC_4(task_team->tt.tt_active)) {
            if (KMP_TASKING_ENABLED(task_team)) {
              int tasks_completed = FALSE;
              __kmp_atomic_execute_tasks_64(
                  this_thr, gtid, (kmp_atomic_flag_64<> *)NULL, FALSE,
                  &tasks_completed USE_ITT_BUILD_ARG(itt_sync_obj), 0);
            } else
              this_thr->th.th_reap_state = KMP_SAFE_TO_REAP;
          }
        } else {
          this_thr->th.th_reap_state = KMP_SAFE_TO_REAP;
        } // if
      }
      if (TCR_4(__kmp_global.g.g_done)) {
        if (__kmp_global.g.g_abort)
          __kmp_abort_thread();
        break;
      } else if (__kmp_tasking_mode != tskm_immediate_exec &&
                 this_thr->th.th_reap_state == KMP_SAFE_TO_REAP) {
        this_thr->th.th_reap_state = KMP_NOT_SAFE_TO_REAP;
      }
    } while (threads_pending > 0);
 
    if (reduce) { // Perform reduction if needed
      OMPT_REDUCTION_DECL(this_thr, gtid);
      OMPT_REDUCTION_BEGIN;
      // Group leader reduces all threads in group
      for (size_t thr = group_start; thr < group_end; thr++) {
        (*reduce)(this_thr->th.th_local.reduce_data,
                  other_threads[thr]->th.th_local.reduce_data);
      }
      OMPT_REDUCTION_END;
    }
 
    // Set flag for next iteration
    b->flags[my_next_iter][tid].stillNeed = 1;
    // Each thread uses a different cache line; resets stillNeed to 0 to
    // indicate it has reached the barrier
    b->flags[my_current_iter][tid].stillNeed = 0;
 
    do { // wait for all group leaders
      threads_pending = 0;
      for (size_t thr = 0; thr < nproc; thr += b->threads_per_group) {
        threads_pending += b->flags[my_current_iter][thr].stillNeed;
      }
      // Execute tasks here
      if (__kmp_tasking_mode != tskm_immediate_exec) {
        kmp_task_team_t *task_team = this_thr->th.th_task_team;
        if (task_team != NULL) {
          if (TCR_SYNC_4(task_team->tt.tt_active)) {
            if (KMP_TASKING_ENABLED(task_team)) {
              int tasks_completed = FALSE;
              __kmp_atomic_execute_tasks_64(
                  this_thr, gtid, (kmp_atomic_flag_64<> *)NULL, FALSE,
                  &tasks_completed USE_ITT_BUILD_ARG(itt_sync_obj), 0);
            } else
              this_thr->th.th_reap_state = KMP_SAFE_TO_REAP;
          }
        } else {
          this_thr->th.th_reap_state = KMP_SAFE_TO_REAP;
        } // if
      }
      if (TCR_4(__kmp_global.g.g_done)) {
        if (__kmp_global.g.g_abort)
          __kmp_abort_thread();
        break;
      } else if (__kmp_tasking_mode != tskm_immediate_exec &&
                 this_thr->th.th_reap_state == KMP_SAFE_TO_REAP) {
        this_thr->th.th_reap_state = KMP_NOT_SAFE_TO_REAP;
      }
    } while (threads_pending > 0);
 
    if (reduce) { // Perform reduction if needed
      if (KMP_MASTER_TID(tid)) { // Master reduces over group leaders
        OMPT_REDUCTION_DECL(this_thr, gtid);
        OMPT_REDUCTION_BEGIN;
        for (size_t thr = b->threads_per_group; thr < nproc;
             thr += b->threads_per_group) {
          (*reduce)(this_thr->th.th_local.reduce_data,
                    other_threads[thr]->th.th_local.reduce_data);
        }
        OMPT_REDUCTION_END;
      }
    }
  } else {
    // Set flag for next iteration
    b->flags[my_next_iter][tid].stillNeed = 1;
    // Each thread uses a different cache line; resets stillNeed to 0 to
    // indicate it has reached the barrier
    b->flags[my_current_iter][tid].stillNeed = 0;
  }
 
  KMP_MFENCE();
 
  KA_TRACE(20,
           ("__kmp_dist_barrier_gather: T#%d(%d:%d) exit for barrier type %d\n",
            gtid, team->t.t_id, tid, bt));
}
 
static void __kmp_dist_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_dist_release);
  kmp_team_t *team;
  distributedBarrier *b;
  kmp_bstate_t *thr_bar;
  kmp_uint64 my_current_iter, next_go;
  size_t my_go_index;
  bool group_leader;
 
  KA_TRACE(20, ("__kmp_dist_barrier_release: T#%d(%d) enter; barrier type %d\n",
                gtid, tid, bt));
 
  thr_bar = &this_thr->th.th_bar[bt].bb;
 
  if (!KMP_MASTER_TID(tid)) {
    // workers and non-master group leaders need to check their presence in team
    do {
      if (this_thr->th.th_used_in_team.load() != 1 &&
          this_thr->th.th_used_in_team.load() != 3) {
        // Thread is not in use in a team. Wait on location in tid's thread
        // struct. The 0 value tells anyone looking that this thread is spinning
        // or sleeping until this location becomes 3 again; 3 is the transition
        // state to get to 1 which is waiting on go and being in the team
        kmp_flag_32<false, false> my_flag(&(this_thr->th.th_used_in_team), 3);
        if (KMP_COMPARE_AND_STORE_ACQ32(&(this_thr->th.th_used_in_team), 2,
                                        0) ||
            this_thr->th.th_used_in_team.load() == 0) {
          my_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
          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_NOTIFan class="preprocessor">#endif
int gtid,

kmp.h:405ass="5
kmp.h:405

before the class="comme382">

@ KMP_IDENT_ATtskm_immediate_exec &aiv class="ttdef">Definition:span) a href="kmp_8h_source.html#l002ss="lineno"> 417 itt_sync_objn class="lineno"> 407 ileeze)( _kmp_itt_barrier_obje="l01662"> gould not class="ne"> 418 n> ">MASKame="l0l00416"> (TCR_SYNC_4(task_tea/span>

&E="lineno"> 375 ">voids.css" rel="stylesheet" type="text/css"/>/sp before the class="commûiv class="line"> * kmp_allol0042/div>