smp.h

Go to the documentation of this file.

#include "tbci/basics.h"

#ifndef TBCI_SMP_H
#define TBCI_SMP_H

#ifdef NO_SMP
# undef SMP
#endif

// libc5 hack 
//#define _POSIX_THREAD_PRIO_INHERIT  

// Check for _POSIX_THREADS, give up otherwise
#if defined(SMP) && defined(HAVE_PTHREADS)

//#warning SMP

#include <signal.h>
#ifndef HAVE_NO_NEW_HEADERS_BUG
# include <cstdlib>
# include <cstdarg>
#else
# include <stdlib.h>
# include <stdarg.h>
#endif
#include <sys/types.h>
#include <sys/mman.h>
#include <limits>
#ifndef PAGESIZE
# define PAGESIZE EXEC_PAGESIZE
#endif
//#include <unistd.h>
//#define __need_sig_atomic_t
#define __need_sigset_t  
#ifdef BUGGY_PTHREAD
# include <sigset.h>
 typedef __sigset_t sigset_t;
#endif
// We need pthread spinlocks (XOPEN2K/XPG6 stuff)
#undef _POSIX_C_SOURCE
#define _POSIX_C_SOURCE 200112L
#include <pthread.h>

#ifndef CACHELINE_SZ
# ifdef __WORDSIZE
#  define CACHELINE_SZ __WORDSIZE
# else
#  warning Setting cache line size manually to 32
#  define CACHELINE_SZ 32
# endif
#endif

NAMESPACE_TBCI

#ifdef __GNUC__
# define smp_barrier() asm ("": : :"memory")
#else
// Oops! No barrier?
# define smp_barrier()  
#endif

extern pid_t main_thread_pid;
#define MAIN_PID (main_thread_pid)
extern bool bound_main;
#endif

#if defined(NEED_SMP_DECLS) || defined(SMP)
#ifdef HAVE_PTHREADS
#include <pthread.h>
#else
#error NEED pthread.h if NEED_SMP_DECLS is set
#endif

typedef void  (*thr_job_t) (struct thr_ctrl*);
typedef void* (*useful_job_t) (void*);

#define THREAD_MAX_ARGS 6
#define THREAD_MAX_RES_LN 16

struct job_input {
        unsigned long t_job_no;
        thr_job_t t_job;                        /* Function pointer */
        unsigned long t_size;                   /* size */
        unsigned long t_off;                    /* offset */
        void* t_par[THREAD_MAX_ARGS];           /* Space for 6 more pointer args */
};
// 80 bytes on 64bit, 40 on 32bit

struct job_output {
        unsigned long t_job_output_no;
        long t_retval;                          /* return value */
        union {
                volatile char t_res_dummy[16];  /* 16 byte space for results */
                LONG_DOUBLE t_res_ld;
                double t_res_d;
                long t_res_l;
                void* t_res_ptr;
        };
};
// 32bytes on 64bit, 24 on 32bit

struct thr_struct {
        int t_no;                                       /* thread_no  */
        pid_t t_pid;                                    /* thread_pid */
        pthread_t t_id;                                 /* thread_id  */
        int t_pipe_to_thread[2], t_pipe_from_thread[2];
        unsigned int t_done_var;
        int numa_node;
} ALIGN(64);

/* For compatibility; this is allocated on the local stack when starting 
 * a thread; input data is copied into it */
struct thr_ctrl {
        unsigned long t_job_no;
        thr_job_t t_job;                        /* Function pointer */
        unsigned long t_size;                   /* size */
        unsigned long t_off;                    /* offset */
        void* t_par[THREAD_MAX_ARGS];           /* Space for 6 more pointer args */
#if 1   
        union {
                volatile char t_res_dummy[16];  /* 16 byte space for results */
                LONG_DOUBLE t_res_ld;
                double t_res_d;
                long t_res_l;
                void* t_res_ptr;
        };
#endif
        int t_no;
};
#endif

#if defined(SMP) && defined(HAVE_PTHREADS)
//extern thr_ctrl master_thr;
extern int num_threads;
extern int threads_busy;
extern int numa_avail;
extern struct thr_struct *threads;
extern THREAD__ int ismainthread;
extern THREAD__ int thrno;
extern THREAD__ struct thr_struct *this_thread;
extern unsigned int curr_n_thr, last_n_thr, prev_n_thr;
#ifdef HAVE_LIBNUMA
extern unsigned page_size;
#endif

int init_threads  (const int thr = 0, const bool load = false);
void free_threads ();
inline int threads_avail (const int wanted = 0)
{
        if (threads_busy)
                return 0; /* avoid deadlock */
        if (wanted > num_threads)
                return num_threads;
        else 
                return wanted;
};

void disable_threads ();
void reenable_threads ();
void bind_threads (bool bind_main = true, bool enable_numa = true, bool add_sibl = false);

//
void thread_start (const int thr_no, thr_job_t job, const unsigned long sz, ...);
void thread_start_off (const int thr_no , thr_job_t job, 
                       const unsigned long offset, const unsigned long sz, ...);
void thread_wait  (const int, job_output* out = 0);
/* wait for a thread to finish and return the double t_res */
double thread_wait_result (const int);

#ifndef SLICE_ALIGN
# define SLICE_ALIGN CACHELINE_SZ
#endif
#ifndef SLICE_DEF_ALIGN
# define SLICE_DEF_ALIGN 8
#endif
/* Helper to slice things into cache aligned tiles */
/* TODO: On NUMA systems, we should better return page size aligned tiles */
template <typename T>
static inline unsigned long slice_offset(int thr, int no_thr, unsigned long dim, T* ptr)
{
#ifdef HAVE_LIBNUMA
        const unsigned int align = numa_avail? page_size: SLICE_ALIGN;
#else
        const unsigned int align = SLICE_ALIGN;
#endif
        if (thr == no_thr)
                return dim;
        else if (thr == 0)
                return 0;
        BCHK(thr < 0 || thr > no_thr, NumErr, Illegal thread in slice_offset, thr, 0);
        const unsigned step = dim/no_thr;
        unsigned long offs = thr*dim/no_thr;
        /* If we don't get a valid pointer, don't try to optimize for cache lines */
        if (!ptr || 2*sizeof(T) > align) {
                offs -= 1;
                return offs + (SLICE_DEF_ALIGN - offs%SLICE_DEF_ALIGN);
        }
        offs += MIN((unsigned int)(align/sizeof(T)), step-1);
        unsigned long misalign = (unsigned long)(ptr+offs)%align / sizeof(T);
        //fprintf(stderr, "Slice %i (base %p): %li(%li) (@%p)\n",
        //         thr, ptr, offs-misalign, offs, ptr+(offs-misalign)); 
        return offs - misalign;
}


inline void update_n_thr(const unsigned int n_thr)
{
        prev_n_thr = last_n_thr; last_n_thr = curr_n_thr; 
        curr_n_thr = n_thr;
}

/* callbacks */
typedef void cbackfn(void *ptr, const int thr);
void thread_reg_callback(cbackfn ctor, cbackfn dtor, void *parm);
void thread_dereg_callback(cbackfn ctor, cbackfn dtor, void *parm);

#ifdef HAVE_LIBNUMA
extern int main_numa_node;
int do_numa_move_pages(int node, int fault_in, 
                       unsigned long firstaddr, unsigned long lastaddr);
void numa_move_pages_job(struct thr_ctrl *tc);
#endif

NAMESPACE_END

#if defined(SMP) && (!defined(_REENTRANT) && !defined(_THREAD_SAFE))
# warning "Define _REENTRANT and/or _THREAD_SAFE for multithreaded (SMP) compilation!"
#endif
        
#else /* _POSIX_THREADS */
# define MAIN_PID (getpid())
# undef SMP
# define num_threads (0)
# define thrno (0)
# define ismainthread (1)
# define threads_avail(x) (0)
static int init_threads(const int=0, const bool=false) { return 0; };
static void bind_threads(bool=true, bool=true, bool=false) {};
static void free_threads () {};
static void disable_threads ()
#ifdef TBCI_OMP
{ omp_set_num_threads(1); }
#else
{};
#endif
static void reenable_threads ()
#ifdef TBCI_OMP
{ omp_set_num_threads(omp_get_num_procs()); }
#else
{};
#endif


//# define numa_optimize(x,y) do {} while(0)
#endif /* _POSIX_THREADS */

#endif /* TBCI_SMP_H */