threadManager_OpenMP.cpp

Go to the documentation of this file.

#include "ResourceManager/threadManager_OpenMP.h"

#include <omp.h>

#include "Communicator/communicator.h"
#include "IO/bridgeIO.h"
using Bridge::vout;

namespace ThreadManager_OpenMP_Reduce {
  template<typename REALTYPE>
  void sum_global(REALTYPE *a,
                  const int num,
                  std::vector<REALTYPE>& array_reduction,
                  const int each_buf_size,
                  const int i_thread, const int Nthread)
  {
    typedef REALTYPE real_t;
    int                 remaining = num;
    std::vector<real_t> sum;
#ifdef LIB_CPP11
    real_t *psum = nullptr;
#else
    real_t *psum = NULL;
#endif

#pragma omp master
    {
      sum.resize(num);
      for (int i = 0; i < num; i++) {
        sum[i] = 0;
      }
      psum = &sum[0];
    }

    real_t *pa = a;
    while (remaining > 0) // sum over threads; shared buffer size is each_buf_Size
    {
      const int n = (remaining < each_buf_size)?num:each_buf_size;
      for (int i = 0; i < n; i++) {
        array_reduction[i_thread * each_buf_size + i] = pa[i];
      }
#pragma omp barrier
#pragma omp master
      {
        for (int i = 0; i < Nthread; i++) {
          for (int j = 0; j < n; j++) {
            psum[j] += array_reduction[i * each_buf_size + j];
          }
        }
        psum += each_buf_size;
      } // master
      pa        += each_buf_size;
      remaining -= each_buf_size;
#pragma omp barrier
    } // sum over threads, done

#pragma omp master
    {
      Communicator::reduce_sum(num, a, &sum[0], 0);
    } // a in the master threads knows the global sum

    remaining = num;
    pa        = a;
    const int total_buf_size = each_buf_size * Nthread;
    while (remaining > 0) // distributes the sum to each thread
    {
      const int n = (remaining < total_buf_size)?num:total_buf_size;
#pragma omp master
      {
        for (int i = 0; i < n; i++) { // copy to the common buffer
          array_reduction[i] = pa[i];
        }
      } // master

      // ensures to read updated m_darray_reduction
#pragma omp barrier
      //#ifdef NECSX
      //#pragma omp flush
      //#else
      //if(sizeof(real_t)==4){
      //#pragma omp flush (ThreadManager_OpenMP::m_darray_reductionF)
      //      } else {
      //#pragma omp flush (ThreadManager_OpenMP::m_darray_reduction)
      // }
      //#endif
      for (int i = 0; i < n; i++) { // copy from the common buffer
        pa[i] = array_reduction[i];
      }
      pa        += total_buf_size;
      remaining -= total_buf_size;
#pragma omp barrier
    } // distributes the sum to each thread, done
  }
}     // ThreadManager_OpenMP_Reduce

//====================================================================
// initialization of static member variables.

int ThreadManager_OpenMP::m_Nthread = 0;
Bridge::VerboseLevel  ThreadManager_OpenMP::m_vl = Bridge::CRUCIAL;
std::vector<dcomplex> ThreadManager_OpenMP::m_darray_reductionDC(0);
std::vector<double>   ThreadManager_OpenMP::m_darray_reduction(0);
std::vector<float>    ThreadManager_OpenMP::m_darray_reductionF(0);

const std::string ThreadManager_OpenMP::class_name = "ThreadManager_OpenMP";

//====================================================================
void ThreadManager_OpenMP::init(int Nthread)
{
  m_vl = CommonParameters::Vlevel();

  vout.general(m_vl, "%s: initialization\n", class_name.c_str());

  int Nthread_env = 0;

#pragma omp parallel
  {
    if (omp_get_thread_num() == 0) {
      Nthread_env = omp_get_num_threads();
    }
  }


  if ((Nthread == Nthread_env) || (Nthread == 0)) {
    m_Nthread = Nthread_env;
  } else {
    vout.general(m_vl, "Warning at %s: Nthread(env) /= Nthread(input)", class_name.c_str());
    vout.general(m_vl, "  Number of threads(env)   = %d\n", Nthread_env);
    vout.general(m_vl, "  Number of threads(input) = %d\n", Nthread);
    vout.general(m_vl, "  reset Nthread = Nthread(input).\n");

    omp_set_num_threads(Nthread);
    m_Nthread = Nthread;
  }

  vout.general(m_vl, "  Number of threads = %d\n", m_Nthread);

  m_darray_reductionDC.resize(m_Nthread * each_buf_sizeDC);
  m_darray_reduction.resize(m_Nthread * each_buf_size);
  m_darray_reductionF.resize(m_Nthread * each_buf_sizeF);
}


//====================================================================
void ThreadManager_OpenMP::finalize()
{
  vout.paranoiac(m_vl, "%s: finalize.\n", class_name.c_str());
}


//====================================================================
int ThreadManager_OpenMP::get_num_threads()
{
  return omp_get_num_threads();
}


//====================================================================
int ThreadManager_OpenMP::get_thread_id()
{
  return omp_get_thread_num();
}


//====================================================================
void ThreadManager_OpenMP::wait()
{
  int Nthread = get_num_threads();

  barrier(Nthread);
}


//====================================================================
void ThreadManager_OpenMP::barrier(int Nthread)
{
#pragma omp barrier
}


//====================================================================
void ThreadManager_OpenMP::sync_barrier_all()
{
#pragma omp barrier
#pragma omp master
  {
    Communicator::sync();
  }
#pragma omp barrier
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(dcomplex& a,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(&a, 1,
                                          m_darray_reductionDC, each_buf_sizeDC,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(dcomplex *a,
                                             const int num,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(a, num,
                                          m_darray_reductionDC, each_buf_sizeDC,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(double& a,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(&a, 1,
                                          m_darray_reduction, each_buf_size,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(double *a,
                                             const int num,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(a, num,
                                          m_darray_reduction, each_buf_size,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(float& a,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(&a, 1,
                                          m_darray_reductionF, each_buf_sizeF,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::reduce_sum_global(float *a,
                                             const int num,
                                             const int i_thread, const int Nthread)
{
  ThreadManager_OpenMP_Reduce::sum_global(a, num,
                                          m_darray_reductionF, each_buf_sizeF,
                                          i_thread, Nthread);
}


//====================================================================
void ThreadManager_OpenMP::assert_single_thread(const std::string& name)
{
  int Nthread = get_num_threads();

  if (Nthread != 1) {
    vout.crucial(m_vl, "\n");
    vout.crucial(m_vl, "##### Caution #####\n");
    vout.crucial(m_vl, "Single-thread %s is called in parallel region.\n", name.c_str());
    vout.crucial(m_vl, "Current number of thread = %d.\n", Nthread);

    exit(EXIT_FAILURE);
  }
}


//====================================================================
//============================================================END=====