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;
  real_t *psum=nullptr;
#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<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, "  Number of threads(env)   = %d\n", Nthread_env);
    vout.general(m_vl, "  Number of threads(input) = %d\n", Nthread);
    vout.general(m_vl, "  resetting Number of threads.\n");
    omp_set_num_threads(Nthread);
    m_Nthread = Nthread;
  }

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

  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(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=====