/Users/mourrain/Devel/mmx/basix/src/threads.cpp

Go to the documentation of this file.

/******************************************************************************
* MODULE     : threads.cpp
* DESCRIPTION: Multi-threading
* COPYRIGHT  : (C) 2007  Joris van der Hoeven
*******************************************************************************
* This software falls under the GNU general public license and comes WITHOUT
* ANY WARRANTY WHATSOEVER. See the file $TEXMACS_PATH/LICENSE for more details.
* If you don't have this file, write to the Free Software Foundation, Inc.,
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
******************************************************************************/

#include <basix/threads.hpp>
#include <basix/port.hpp>
#ifdef __linux__
#include <unistd.h>
#else
#  ifdef __APPLE__
#  include <sys/types.h>
#  include <sys/sysctl.h>
#  endif
#endif

namespace mmx {

mutex memory_lock;

#ifdef BASIX_ENABLE_THREADS

/******************************************************************************
* Global data
******************************************************************************/

struct thread_data {
  pthread_t th;
  nat       id;
  task_rep* t;
};

bool            threads_busy     = false;
bool            threads_active   = false;
nat             threads_number   = 0;
thread_data*    threads_data     = NULL;

task*           threads_task     = NULL;
nat             threads_nr_tasks = 0;
nat             threads_index    = 0;
nat             threads_todo     = 0;

pthread_mutex_t threads_task_lock;
pthread_mutex_t threads_idle_lock;
pthread_cond_t  threads_idle_cond;
pthread_mutex_t threads_busy_lock;
pthread_cond_t  threads_busy_cond;

/******************************************************************************
* Workers
******************************************************************************/

void*
threads_worker (void* ptr) {
  thread_data* data= (thread_data*) ptr;
  //pthread_mutex_lock (&threads_busy_lock);
  //mmout << "Started thread #" << data->id << "\n";
  //pthread_mutex_unlock (&threads_busy_lock);

  while (threads_busy) {
    // Wait for new input
    pthread_mutex_lock (&threads_idle_lock);
    //mmout << "Thread #" << data->id << " is waiting\n";
    pthread_cond_wait (&threads_idle_cond, &threads_idle_lock);
    pthread_mutex_unlock (&threads_idle_lock);
    if (!threads_busy) break;

    while (true) {
      // Retrieve next task
      pthread_mutex_lock (&threads_task_lock);
      if (threads_index == threads_nr_tasks) {
        pthread_mutex_unlock (&threads_task_lock);
        break;
      }
      //mmout << "Thread #" << data->id
      //      << " processes task " << threads_index << "\n";
      data->t= inside (threads_task[threads_index]);
      threads_index++;
      pthread_mutex_unlock (&threads_task_lock);

      // Execute next task
      if (data->t != NULL) data->t->execute ();

      // Remove task
      data->t= NULL;
      pthread_mutex_lock (&threads_busy_lock);
      threads_todo--;
      //mmout << "Thread #" << data->id << " completed; "
      //      << threads_todo << " tasks waiting\n";
      if (threads_todo == 0)
        pthread_cond_signal (&threads_busy_cond);
      pthread_mutex_unlock (&threads_busy_lock);
    }
  }

  pthread_mutex_lock (&threads_busy_lock);
  threads_todo--;
  //mmout << "Thread #" << data->id << " dying; "
  //      << threads_todo << " tasks waiting to die\n";
  if (threads_todo == 0)
    pthread_cond_signal (&threads_busy_cond);
  pthread_mutex_unlock (&threads_busy_lock);
  pthread_exit (NULL);
}

/******************************************************************************
* Master
******************************************************************************/

void
threads_execute (task* ts, nat n) {
  threads_task    = ts;
  threads_nr_tasks= n;
  threads_index   = 0;
  threads_todo    = n;

  threads_active  = true;
  pthread_mutex_lock (&threads_busy_lock);
  pthread_mutex_lock (&threads_idle_lock);
  pthread_cond_broadcast (&threads_idle_cond);
  pthread_mutex_unlock (&threads_idle_lock);

  pthread_cond_wait (&threads_busy_cond, &threads_busy_lock);
  pthread_mutex_unlock (&threads_busy_lock);
  threads_active  = false;
}

/******************************************************************************
* Initialization and termination
******************************************************************************/

void
threads_initialize (nat n) {
  pthread_mutex_init (&threads_task_lock, NULL);
  pthread_mutex_init (&threads_idle_lock, NULL);
  pthread_cond_init  (&threads_idle_cond, NULL);
  pthread_mutex_init (&threads_busy_lock, NULL);
  pthread_cond_init  (&threads_busy_cond, NULL);

  threads_busy   = true;
  threads_number = n;
  threads_data   = new thread_data[threads_number];
  for (nat id=0; id<threads_number; id++) {
    //pthread_attr_t attr;
    //pthread_attr_init(&attr);
    //pthread_attr_setscope (&attr, PTHREAD_SCOPE_SYSTEM);
    threads_data[id].id= id;
    threads_data[id].t = NULL;
    if (pthread_create(&(threads_data[id].th), NULL /* &attr */,
                       threads_worker, (void *) (threads_data + id)))
      ERROR ("failed to create thread");
  }
}

void
threads_terminate () {
  threads_busy    = false;
  threads_todo    = threads_number;

  pthread_mutex_lock (&threads_busy_lock);
  pthread_mutex_lock (&threads_idle_lock);
  pthread_cond_broadcast (&threads_idle_cond);
  pthread_mutex_unlock (&threads_idle_lock);

  pthread_cond_wait (&threads_busy_cond, &threads_busy_lock);
  pthread_mutex_unlock (&threads_busy_lock);

  void* exit_status;
  for (nat id=0; id<threads_number; id++)
    pthread_join (threads_data[id].th, &exit_status);
  
  delete[] threads_data;
  threads_data   = NULL;
  threads_number = 0;

  pthread_mutex_destroy (&threads_task_lock);
  pthread_mutex_destroy (&threads_idle_lock);
  pthread_cond_destroy  (&threads_idle_cond);
  pthread_mutex_destroy (&threads_busy_lock);
  pthread_cond_destroy  (&threads_busy_cond);
}

struct threads_instance {
  threads_instance (nat n) {
    threads_initialize (n); }
  ~threads_instance () {
    threads_terminate (); }
};

threads_instance threads_inst (BASIX_ENABLE_THREADS);

int
threads_get_number () {
  return threads_number;
}

void
threads_set_number (const int& n) {
  threads_terminate ();
  threads_initialize (n);
}

/******************************************************************************
* Multi-threading disabled
******************************************************************************/

#else // ifndef BASIX_ENABLE_THREADS

nat  threads_number= 1;
void threads_initialize (nat n) { (void) n; }
void threads_terminate () {}
int threads_get_number () { return 1; }
void threads_set_number (const int& n) {
  ASSERT (n == 1, "multithreading disabled"); }
void init_threads () {}
void term_threads () {}
void threads_execute (task* ts, nat n) {
  for (nat i=0; i<n; i++) inside (ts[i]) -> execute (); }

#endif // BASIX_ENABLE_THREADS

/******************************************************************************
* Platform independent determination of available number of cores
******************************************************************************/

int
get_number_cores () {
#ifdef __linux__ // also to be used for Solaris, AIX, ...
  int nr1= sysconf (_SC_NPROCESSORS_ONLN);
  if (nr1 >= 1) return nr1;
#endif

#ifdef __APPLE__ // also to be used for BSD system
  int mib[2] = { CTL_HW, HW_AVAILCPU };
  //int mib[2] = { CTL_HW, HW_NCPU };
  int nr2;
  size_t len;
  len = sizeof (nr2);
  sysctl (mib, 2, &nr2, &len, NULL, 0);
  if (nr2 >= 1) return nr2;
#endif

#ifdef __WINDOW__upcoming
  SYSTEM_INFO sysinfo;
  GetSystemInfo( &sysinfo );
  int nr3= sysinfo.dwNumberOfProcessors;
  if (nr3 >= 1) return nr3;
#endif

#ifdef __HPUX__upcoming
  int nr4= mpctl (MPC_GETNUMSPUS, NULL, NULL);
  if (nr4 >= 1) return nr4;
#endif

#ifdef __IRIX__upcoming
  int nr5= sysconf (_SC_NPROC_ONLN);
  if (nr5 >= 1) return nr5;
#endif

  return 1;
}

/******************************************************************************
* Running a stupid loop for testing purposes
******************************************************************************/

struct loop_task_rep: public task_rep {
  nat iterations;
public:
  inline loop_task_rep (nat its): iterations (its) {}
  void execute () { for (nat i=0; i<iterations; i++) noop (); }
};

void
threads_simple_loop (nat nr_instances, nat nr_iterations) {
  task ts[nr_instances];
  for (nat id=0; id<nr_instances; id++)
    ts[id]= new loop_task_rep (nr_iterations);
  threads_execute (ts, nr_instances);
}

} // namespace mmx