cannam@95: /*
cannam@95:  * Copyright (c) 2003, 2007-11 Matteo Frigo
cannam@95:  * Copyright (c) 2003, 2007-11 Massachusetts Institute of Technology
cannam@95:  *
cannam@95:  * This program is free software; you can redistribute it and/or modify
cannam@95:  * it under the terms of the GNU General Public License as published by
cannam@95:  * the Free Software Foundation; either version 2 of the License, or
cannam@95:  * (at your option) any later version.
cannam@95:  *
cannam@95:  * This program is distributed in the hope that it will be useful,
cannam@95:  * but WITHOUT ANY WARRANTY; without even the implied warranty of
cannam@95:  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
cannam@95:  * GNU General Public License for more details.
cannam@95:  *
cannam@95:  * You should have received a copy of the GNU General Public License
cannam@95:  * along with this program; if not, write to the Free Software
cannam@95:  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
cannam@95:  *
cannam@95:  */
cannam@95: 
cannam@95: /* threads.c: Portable thread spawning for loops, via the X(spawn_loop)
cannam@95:    function.  The first portion of this file is a set of macros to
cannam@95:    spawn and join threads on various systems. */
cannam@95: 
cannam@95: #include "threads.h"
cannam@95: 
cannam@95: #if defined(USING_POSIX_THREADS)
cannam@95: 
cannam@95: #include <pthread.h>
cannam@95: 
cannam@95: #ifdef HAVE_UNISTD_H
cannam@95: #  include <unistd.h>
cannam@95: #endif
cannam@95: 
cannam@95: /* imlementation of semaphores and mutexes: */
cannam@95: #if (defined(_POSIX_SEMAPHORES) && (_POSIX_SEMAPHORES >= 200112L))
cannam@95: 
cannam@95:    /* If optional POSIX semaphores are supported, use them to
cannam@95:       implement both semaphores and mutexes. */
cannam@95: #  include <semaphore.h>
cannam@95: #  include <errno.h>
cannam@95: 
cannam@95:    typedef sem_t os_sem_t;
cannam@95: 
cannam@95:    static void os_sem_init(os_sem_t *s) { sem_init(s, 0, 0); }
cannam@95:    static void os_sem_destroy(os_sem_t *s) { sem_destroy(s); }
cannam@95: 
cannam@95:    static void os_sem_down(os_sem_t *s)
cannam@95:    {
cannam@95: 	int err;
cannam@95: 	do {
cannam@95: 	     err = sem_wait(s);
cannam@95: 	} while (err == -1 && errno == EINTR);
cannam@95: 	CK(err == 0);
cannam@95:    }
cannam@95: 
cannam@95:    static void os_sem_up(os_sem_t *s) { sem_post(s); }
cannam@95: 
cannam@95:    /*
cannam@95:       The reason why we use sem_t to implement mutexes is that I have
cannam@95:       seen mysterious hangs with glibc-2.7 and linux-2.6.22 when using
cannam@95:       pthread_mutex_t, but no hangs with sem_t or with linux >=
cannam@95:       2.6.24.  For lack of better information, sem_t looks like the
cannam@95:       safest choice.
cannam@95:    */
cannam@95:    typedef sem_t os_mutex_t;
cannam@95:    static void os_mutex_init(os_mutex_t *s) { sem_init(s, 0, 1); }
cannam@95:    #define os_mutex_destroy os_sem_destroy
cannam@95:    #define os_mutex_lock os_sem_down
cannam@95:    #define os_mutex_unlock os_sem_up
cannam@95: 
cannam@95: #else
cannam@95: 
cannam@95:    /* If optional POSIX semaphores are not defined, use pthread
cannam@95:       mutexes for mutexes, and simulate semaphores with condition
cannam@95:       variables */
cannam@95:    typedef pthread_mutex_t os_mutex_t;
cannam@95: 
cannam@95:    static void os_mutex_init(os_mutex_t *s) 
cannam@95:    { 
cannam@95: 	pthread_mutex_init(s, (pthread_mutexattr_t *)0);
cannam@95:    }
cannam@95: 
cannam@95:    static void os_mutex_destroy(os_mutex_t *s) { pthread_mutex_destroy(s); }
cannam@95:    static void os_mutex_lock(os_mutex_t *s) { pthread_mutex_lock(s); }
cannam@95:    static void os_mutex_unlock(os_mutex_t *s) { pthread_mutex_unlock(s); }
cannam@95: 
cannam@95:    typedef struct {
cannam@95: 	pthread_mutex_t m;
cannam@95: 	pthread_cond_t c;
cannam@95: 	volatile int x;
cannam@95:    } os_sem_t; 
cannam@95: 
cannam@95:    static void os_sem_init(os_sem_t *s)
cannam@95:    {
cannam@95: 	pthread_mutex_init(&s->m, (pthread_mutexattr_t *)0);
cannam@95: 	pthread_cond_init(&s->c, (pthread_condattr_t *)0);
cannam@95: 
cannam@95: 	/* wrap initialization in lock to exploit the release
cannam@95: 	   semantics of pthread_mutex_unlock() */
cannam@95: 	pthread_mutex_lock(&s->m);
cannam@95: 	s->x = 0;
cannam@95: 	pthread_mutex_unlock(&s->m);
cannam@95:    }
cannam@95: 
cannam@95:    static void os_sem_destroy(os_sem_t *s)
cannam@95:    {
cannam@95: 	pthread_mutex_destroy(&s->m);
cannam@95: 	pthread_cond_destroy(&s->c);
cannam@95:    }
cannam@95: 
cannam@95:    static void os_sem_down(os_sem_t *s)
cannam@95:    {
cannam@95: 	pthread_mutex_lock(&s->m);
cannam@95: 	while (s->x <= 0) 
cannam@95: 	     pthread_cond_wait(&s->c, &s->m);
cannam@95: 	--s->x;
cannam@95: 	pthread_mutex_unlock(&s->m);
cannam@95:    }
cannam@95: 
cannam@95:    static void os_sem_up(os_sem_t *s)
cannam@95:    {
cannam@95: 	pthread_mutex_lock(&s->m);
cannam@95: 	++s->x;
cannam@95: 	pthread_cond_signal(&s->c);
cannam@95: 	pthread_mutex_unlock(&s->m);
cannam@95:    }
cannam@95: 
cannam@95: #endif
cannam@95: 
cannam@95: #define FFTW_WORKER void *
cannam@95: 
cannam@95: static void os_create_thread(FFTW_WORKER (*worker)(void *arg), 
cannam@95: 			     void *arg)
cannam@95: {
cannam@95:      pthread_attr_t attr;
cannam@95:      pthread_t tid;
cannam@95: 
cannam@95:      pthread_attr_init(&attr);
cannam@95:      pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM); 
cannam@95:      pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
cannam@95: 
cannam@95:      pthread_create(&tid, &attr, worker, (void *)arg);
cannam@95:      pthread_attr_destroy(&attr);
cannam@95: }
cannam@95: 
cannam@95: static void os_destroy_thread(void)
cannam@95: {
cannam@95:      pthread_exit((void *)0);
cannam@95: }
cannam@95: 
cannam@95: #elif defined(__WIN32__) || defined(_WIN32) || defined(_WINDOWS)
cannam@95: /* hack: windef.h defines INT for its own purposes and this causes
cannam@95:    a conflict with our own INT in ifftw.h.  Divert the windows
cannam@95:    definition into another name unlikely to cause a conflict */
cannam@95: #define INT magnus_ab_INTegro_seclorum_nascitur_ordo
cannam@95: #include <windows.h>
cannam@95: #include <process.h>
cannam@95: #undef INT
cannam@95: 
cannam@95: typedef HANDLE os_mutex_t;
cannam@95: 
cannam@95: static void os_mutex_init(os_mutex_t *s) 
cannam@95: { 
cannam@95:      *s = CreateMutex(NULL, FALSE, NULL);
cannam@95: }
cannam@95: 
cannam@95: static void os_mutex_destroy(os_mutex_t *s) 
cannam@95: { 
cannam@95:      CloseHandle(*s);
cannam@95: }
cannam@95: 
cannam@95: static void os_mutex_lock(os_mutex_t *s)
cannam@95: { 
cannam@95:      WaitForSingleObject(*s, INFINITE);
cannam@95: }
cannam@95: 
cannam@95: static void os_mutex_unlock(os_mutex_t *s) 
cannam@95: { 
cannam@95:      ReleaseMutex(*s);
cannam@95: }
cannam@95: 
cannam@95: typedef HANDLE os_sem_t;
cannam@95: 
cannam@95: static void os_sem_init(os_sem_t *s) 
cannam@95: {
cannam@95:      *s = CreateSemaphore(NULL, 0, 0x7FFFFFFFL, NULL);
cannam@95: }
cannam@95: 
cannam@95: static void os_sem_destroy(os_sem_t *s) 
cannam@95: { 
cannam@95:      CloseHandle(*s);
cannam@95: }
cannam@95: 
cannam@95: static void os_sem_down(os_sem_t *s) 
cannam@95: { 
cannam@95:      WaitForSingleObject(*s, INFINITE);
cannam@95: }
cannam@95: 
cannam@95: static void os_sem_up(os_sem_t *s) 
cannam@95: {
cannam@95:      ReleaseSemaphore(*s, 1, NULL);
cannam@95: }
cannam@95: 
cannam@95: #define FFTW_WORKER unsigned __stdcall
cannam@95: typedef unsigned (__stdcall *winthread_start) (void *);
cannam@95: 
cannam@95: static void os_create_thread(winthread_start worker,
cannam@95: 			     void *arg)
cannam@95: {
cannam@95:      _beginthreadex((void *)NULL,               /* security attrib */
cannam@95: 		    0,				/* stack size */
cannam@95: 		    worker,                     /* start address */
cannam@95: 		    arg,			/* parameters */
cannam@95: 		    0,				/* creation flags */
cannam@95: 		    (unsigned *)NULL);		/* tid */
cannam@95: }
cannam@95: 
cannam@95: static void os_destroy_thread(void)
cannam@95: {
cannam@95:      _endthreadex(0);
cannam@95: }
cannam@95: 
cannam@95: 
cannam@95: #else
cannam@95: #error "No threading layer defined"
cannam@95: #endif
cannam@95: 
cannam@95: /************************************************************************/
cannam@95: 
cannam@95: /* Main code: */
cannam@95: struct worker {
cannam@95:      os_sem_t ready;
cannam@95:      os_sem_t done;
cannam@95:      struct work *w;
cannam@95:      struct worker *cdr;
cannam@95: };
cannam@95: 
cannam@95: static struct worker *make_worker(void)
cannam@95: {
cannam@95:      struct worker *q = (struct worker *)MALLOC(sizeof(*q), OTHER);
cannam@95:      os_sem_init(&q->ready);
cannam@95:      os_sem_init(&q->done);
cannam@95:      return q;
cannam@95: }
cannam@95: 
cannam@95: static void unmake_worker(struct worker *q)
cannam@95: {
cannam@95:      os_sem_destroy(&q->done);
cannam@95:      os_sem_destroy(&q->ready);
cannam@95:      X(ifree)(q);
cannam@95: }
cannam@95: 
cannam@95: struct work {
cannam@95:      spawn_function proc;
cannam@95:      spawn_data d;
cannam@95:      struct worker *q; /* the worker responsible for performing this work */
cannam@95: };
cannam@95: 
cannam@95: static os_mutex_t queue_lock;
cannam@95: static os_sem_t termination_semaphore;
cannam@95: 
cannam@95: static struct worker *worker_queue;
cannam@95: #define WITH_QUEUE_LOCK(what)			\
cannam@95: {						\
cannam@95:      os_mutex_lock(&queue_lock);		\
cannam@95:      what;					\
cannam@95:      os_mutex_unlock(&queue_lock);		\
cannam@95: }
cannam@95: 
cannam@95: static FFTW_WORKER worker(void *arg)
cannam@95: {
cannam@95:      struct worker *ego = (struct worker *)arg;
cannam@95:      struct work *w;
cannam@95: 
cannam@95:      for (;;) {
cannam@95: 	  /* wait until work becomes available */
cannam@95: 	  os_sem_down(&ego->ready);
cannam@95: 
cannam@95: 	  w = ego->w;
cannam@95: 
cannam@95: 	  /* !w->proc ==> terminate worker */
cannam@95: 	  if (!w->proc) break;
cannam@95: 
cannam@95: 	  /* do the work */
cannam@95:           w->proc(&w->d);
cannam@95: 
cannam@95: 	  /* signal that work is done */
cannam@95: 	  os_sem_up(&ego->done);
cannam@95:      }
cannam@95: 
cannam@95:      /* termination protocol */
cannam@95:      os_sem_up(&termination_semaphore);
cannam@95: 
cannam@95:      os_destroy_thread();
cannam@95:      /* UNREACHABLE */
cannam@95:      return 0;
cannam@95: }
cannam@95: 
cannam@95: static void enqueue(struct worker *q)
cannam@95: {
cannam@95:      WITH_QUEUE_LOCK({
cannam@95: 	  q->cdr = worker_queue;
cannam@95: 	  worker_queue = q;
cannam@95:      });
cannam@95: }
cannam@95: 
cannam@95: static struct worker *dequeue(void)
cannam@95: {
cannam@95:      struct worker *q;
cannam@95: 
cannam@95:      WITH_QUEUE_LOCK({
cannam@95: 	  q = worker_queue;
cannam@95: 	  if (q) 
cannam@95: 	       worker_queue = q->cdr;
cannam@95:      });
cannam@95: 
cannam@95:      if (!q) {
cannam@95: 	  /* no worker is available.  Create one */
cannam@95: 	  q = make_worker();
cannam@95: 	  os_create_thread(worker, q);
cannam@95:      }
cannam@95: 
cannam@95:      return q;
cannam@95: }
cannam@95: 
cannam@95: 
cannam@95: static void kill_workforce(void)
cannam@95: {
cannam@95:      struct work w;
cannam@95: 
cannam@95:      w.proc = 0;
cannam@95: 
cannam@95:      THREAD_ON; /* needed for debugging mode: since make_worker
cannam@95: 		   is called from dequeue which is only called in
cannam@95: 		   thread_on mode, we need to unmake_worker in thread_on. */
cannam@95:      WITH_QUEUE_LOCK({
cannam@95: 	  /* tell all workers that they must terminate.  
cannam@95: 
cannam@95: 	     Because workers enqueue themselves before signaling the
cannam@95: 	     completion of the work, all workers belong to the worker queue
cannam@95: 	     if we get here.  Also, all workers are waiting at
cannam@95: 	     os_sem_down(ready), so we can hold the queue lock without
cannam@95: 	     deadlocking */
cannam@95: 	  while (worker_queue) {
cannam@95: 	       struct worker *q = worker_queue;
cannam@95: 	       worker_queue = q->cdr;
cannam@95: 	       q->w = &w;
cannam@95: 	       os_sem_up(&q->ready);
cannam@95: 	       os_sem_down(&termination_semaphore);
cannam@95: 	       unmake_worker(q);
cannam@95: 	  }
cannam@95:      });
cannam@95:      THREAD_OFF;
cannam@95: }
cannam@95: 
cannam@95: int X(ithreads_init)(void)
cannam@95: {
cannam@95:      os_mutex_init(&queue_lock);
cannam@95:      os_sem_init(&termination_semaphore);
cannam@95: 
cannam@95:      WITH_QUEUE_LOCK({
cannam@95: 	  worker_queue = 0;
cannam@95:      })
cannam@95: 
cannam@95:      return 0; /* no error */
cannam@95: }
cannam@95: 
cannam@95: /* Distribute a loop from 0 to loopmax-1 over nthreads threads.
cannam@95:    proc(d) is called to execute a block of iterations from d->min
cannam@95:    to d->max-1.  d->thr_num indicate the number of the thread
cannam@95:    that is executing proc (from 0 to nthreads-1), and d->data is
cannam@95:    the same as the data parameter passed to X(spawn_loop).
cannam@95: 
cannam@95:    This function returns only after all the threads have completed. */
cannam@95: void X(spawn_loop)(int loopmax, int nthr, spawn_function proc, void *data)
cannam@95: {
cannam@95:      int block_size;
cannam@95:      struct work *r;
cannam@95:      int i;
cannam@95: 
cannam@95:      A(loopmax >= 0);
cannam@95:      A(nthr > 0);
cannam@95:      A(proc);
cannam@95: 
cannam@95:      if (!loopmax) return;
cannam@95: 
cannam@95:      /* Choose the block size and number of threads in order to (1)
cannam@95:         minimize the critical path and (2) use the fewest threads that
cannam@95:         achieve the same critical path (to minimize overhead).
cannam@95:         e.g. if loopmax is 5 and nthr is 4, we should use only 3
cannam@95:         threads with block sizes of 2, 2, and 1. */
cannam@95:      block_size = (loopmax + nthr - 1) / nthr;
cannam@95:      nthr = (loopmax + block_size - 1) / block_size;
cannam@95: 
cannam@95:      THREAD_ON; /* prevent debugging mode from failing under threads */
cannam@95:      STACK_MALLOC(struct work *, r, sizeof(struct work) * nthr);
cannam@95: 	  
cannam@95:      /* distribute work: */
cannam@95:      for (i = 0; i < nthr; ++i) {
cannam@95: 	  struct work *w = &r[i];
cannam@95: 	  spawn_data *d = &w->d;
cannam@95: 
cannam@95: 	  d->max = (d->min = i * block_size) + block_size;
cannam@95: 	  if (d->max > loopmax)
cannam@95: 	       d->max = loopmax;
cannam@95: 	  d->thr_num = i;
cannam@95: 	  d->data = data;
cannam@95: 	  w->proc = proc;
cannam@95: 	   
cannam@95: 	  if (i == nthr - 1) {
cannam@95: 	       /* do the work ourselves */
cannam@95: 	       proc(d);
cannam@95: 	  } else {
cannam@95: 	       /* assign a worker to W */
cannam@95: 	       w->q = dequeue();
cannam@95: 
cannam@95: 	       /* tell worker w->q to do it */
cannam@95: 	       w->q->w = w; /* Dirac could have written this */
cannam@95: 	       os_sem_up(&w->q->ready);
cannam@95: 	  }
cannam@95:      }
cannam@95: 
cannam@95:      for (i = 0; i < nthr - 1; ++i) { 
cannam@95: 	  struct work *w = &r[i];
cannam@95: 	  os_sem_down(&w->q->done);
cannam@95: 	  enqueue(w->q);
cannam@95:      }
cannam@95: 
cannam@95:      STACK_FREE(r);
cannam@95:      THREAD_OFF; /* prevent debugging mode from failing under threads */
cannam@95: }
cannam@95: 
cannam@95: void X(threads_cleanup)(void)
cannam@95: {
cannam@95:      kill_workforce();
cannam@95:      os_mutex_destroy(&queue_lock);
cannam@95:      os_sem_destroy(&termination_semaphore);
cannam@95: }