smp.cc

Go to the documentation of this file.

#define _GNU_SOURCE 1

#define NEED_SMP_DECLS 1
//#include "basics.h"
#include "smp.h"
#include <stdio.h>
#include <time.h>

#include <errno.h>

#ifdef _POSIX_THREADS
#ifdef SMP

//#include <signum.h>
#include <stdlib.h>

#ifdef HAVE_SYS_SYSINFO_H
# include <sys/sysinfo.h>
#endif

#ifdef HAVE_SCHED_H
# include <sched.h>
#endif

#ifndef MAX_THREADS
# define MAX_THREADS 160
#endif

#include <list.h>

NAMESPACE_TBCI

int num_threads = 0;
int threads_busy = 0;
struct thr_ctrl *threads = 0;
pid_t main_thread_pid = 0;
bool threads_bound = false;
bool bound_main = false;
THREAD__ int ismainthread = 1;
THREAD__ int thrno = 0;
THREAD__ struct thr_ctrl *this_thread = 0;

#ifdef DEBUG_THREAD
# define TCHK(x) if (UNLIKELY(err = x)) fprintf (stderr, #x " failed: %s\n", CSTD__ strerror (err))
# define ERRDECL int err = 0
#else
# define TCHK(x) x
# define ERRDECL 
#endif

#ifdef HAVE_SCHED_GETAFFINITY
static cpu_set_t saved_cpuset;

#ifndef HAVE_CPU_COUNT
#undef CPU_COUNT
static int CPU_COUNT(const cpu_set_t *cpus)
{
        int cnt = 0;
        for (int i = 0; i < CPU_SETSIZE; ++i)
                if (CPU_ISSET(i, cpus))
                        ++cnt; 
        return cnt;
}
#endif
#ifndef CPU_XOR
static void CPU_XOR(cpu_set_t *dest, cpu_set_t *src1, cpu_set_t *src2)
{
        for (int i = 0; i < CPU_SETSIZE; ++i)
                if (CPU_ISSET(i, src1) ^ CPU_ISSET(i, src2))
                        CPU_SET(i, dest);
                else
                        CPU_CLR(i, dest);
}
#endif
#ifndef CPU_AND
static void CPU_AND(cpu_set_t *dest, cpu_set_t *src1, cpu_set_t *src2)
{
        for (int i = 0; i < CPU_SETSIZE; ++i)
                if (CPU_ISSET(i, src1) & CPU_ISSET(i, src2))
                        CPU_SET(i, dest);
                else
                        CPU_CLR(i, dest);
}
#endif


static int next_set_bit(const int start, const cpu_set_t *cpus)
{
        for (int i = start; i < CPU_SETSIZE; ++i)
                if (CPU_ISSET(i, cpus))
                        return i;
        return -1;
}

static char cpu_buf[512];
static const char* cpu_str(const cpu_set_t *cpus)
{
        int ptr = 0; cpu_buf[1] = 0;
        for (int i = 0; i < CPU_SETSIZE && ptr < 504; ++i)
                if (CPU_ISSET(i, cpus))
                        ptr += sprintf(cpu_buf+ptr, " %i", i);
        return cpu_buf+1;
}
#define CPU_SETBYTES ((CPU_SETSIZE+7)/8)
#if defined(__linux__)

static void parse_and_remove_siblings(int cpu, cpu_set_t *cpus)
{
        char fn[80];
        sprintf(fn, "/sys/devices/system/cpu/cpu%i/topology/thread_siblings", cpu);
        FILE *f = fopen(fn, "r");
        if (!f)
                return;
        int sibl[(CPU_SETSIZE+31)/32]; 
        int nr_ints = 0;
        bool end = false;
        /* Linux reports this as field of ints, last int covers first 32 CPUs */
        while (!end) {
                int msk;
                if (fscanf(f, "%x,", &msk) == EOF) {
                        if (fscanf(f, "%x", &msk) == EOF)
                                break;
                        end = true;
                }
                sibl[nr_ints++] = msk;
        }
        fclose(f);
        /* Convert to cpu_set_t data structure (machine endian field of longs)
         * first long covers first 32/64 CPUs */
        cpu_set_t sibl_set; CPU_ZERO(&sibl_set);
        long* sibl_ptr = (long*)&sibl_set;
#if __BYTE_ORDER == __LITTLE_ENDIAN
        for (int j = nr_ints-1; j >= 0; --j)
                *sibl_ptr++ = sibl[j];
#else
        long* sibl_ptr2 = (long*)&sibl;
        for (int j = (nr_ints*sizeof(int)/sizeof(long))-1; j >= 0; --j)
                *sibl_ptr++ = sibl_ptr2[j];
#endif
        //fprintf(stderr, "Siblings for cpu%i: %s\n", cpu, cpu_str(&sibl_set));
        //fprintf(stderr, "CPUS before rmv: %s\n", cpu_str(cpus));
        if (!CPU_ISSET(cpu, &sibl_set))
                abort();
        CPU_CLR(cpu, &sibl_set);
        CPU_AND(&sibl_set, &sibl_set, cpus);
        CPU_XOR(cpus, cpus, &sibl_set);
        //fprintf(stderr, "CPUS after rmvl: %s\n", cpu_str(cpus));
}

static void remove_hyperthreads(cpu_set_t *cpus)
{
#ifdef THREAD_STAT
        fprintf(stderr, "CPU set before HT removal: %s\n", cpu_str(cpus));
#endif
        for (int i = CPU_SETSIZE-1; i >=0; --i) {
                if (!CPU_ISSET(i, cpus))
                        continue;
                parse_and_remove_siblings(i, cpus);
        }
#ifdef THREAD_STAT
        fprintf(stderr, "CPU set after  HT removal: %s\n", cpu_str(cpus));
#endif
}
#else
static void remove_hyperthreads(cpu_set_t *cpus)
{
}
#endif

#endif

/* detect the # of cpus for glibc and/or linux systems */
static int detect_num_cpu ()
{
        int cpus = 0;
#ifdef HAVE_SCHED_GETAFFINITY
        pid_t pid = getpid();
        if (! sched_getaffinity(pid, CPU_SETBYTES, &saved_cpuset)) {
                remove_hyperthreads(&saved_cpuset);
                return CPU_COUNT(&saved_cpuset);
        }
#endif
#ifdef HAVE_GET_NPROCS // defined(__GLIBC__) &&  __GLIBC__ >= 2
        cpus = get_nprocs ();
        if (cpus <= 0) 
                return 2;
#elif defined(__linux__) && (defined(__i386__) || defined(__x86_64__))
        FILE* cpuinfo;
        char buf[128];
        cpuinfo = fopen ("/proc/cpuinfo", "r");
        if (cpuinfo <= 0) 
                return 2;       /* return 2, if we can't find out */
        while (!feof (cpuinfo)) {
                fgets (buf, 128, cpuinfo);
                if (CSTD__ memcmp (buf+1, "rocessor", 8) == 0) 
                        cpus++;
        }
#endif
#ifdef DEBUG_THREAD
        fprintf (stderr, "%i CPUs detected.\n", cpus);
#endif
        return (cpus == 0 ? 1: cpus);
}

#ifdef HAVE_GETLOADAVG
static int loadavg ()
{
        double loads[3];
        int err = getloadavg (loads, 3);
        if (err == 1)
                return (int)loads[0];
        else if (err > 1)
                return (int)loads[1];
        else 
                return 0;
}
#endif

class cback {
        public:
                cbackfn *ctor;
                cbackfn *dtor;
                void *parm;
                cback(cbackfn ct, cbackfn dt, void *p)
                : ctor(ct), dtor(dt), parm(p) {};
                cback()
                : ctor(NULL), dtor(NULL), parm(NULL) {};
                void init(const int thr)
                { ctor(parm, thr); }
                void deinit(const int thr)
                { dtor(parm, thr); }
};

static List<cback> thread_cbacks;
/* Instantiate */
template class List<cback>;

void thread_reg_callback(cbackfn ctor, cbackfn dtor, void *parm)
{
        cback callback(ctor, dtor, parm);
        thread_cbacks.append(&callback);
        //fprintf(stderr, "Register callback %p (%p %p %p)\n", &callback, ctor, dtor, parm);
}

void thread_dereg_callback(cbackfn ctor, cbackfn dtor, void *parm)
{
        cback callback(ctor, dtor, parm);
        cback *cbk = thread_cbacks.setcurr(&callback);
        BCHK(!cbk, NumErr, Deregister unrgistered callback, (long int)ctor, );
        thread_cbacks.delcurr();
        //fprintf(stderr, "Deregister callback %p (%p %p %p)\n", &callback, ctor, dtor, parm);
}

#if defined(__i386__) || defined(__x86_64__)
# define _cpu_relax() asm ("rep; nop")
#else
# define _cpu_relax()
#endif

unsigned long poll_succ = 0;
unsigned long poll_fail = 0;
unsigned long polls_succ = 0;
unsigned long polls_fail = 0;

#ifndef POLL_REP
# define POLL_REP 320
#endif
#ifndef POLL_REPS
# define POLL_REPS (POLL_REP/16)
#endif

/* helper, the opposite of pthr_cond_wait (cond, mut) */
inline void pthr_cond_signal_mutex (pthread_cond_t *cond, pthread_mutex_t *mut, volatile int *done)
{
        ERRDECL;
        int w = 0;
        *done = 1;
        TCHK(pthread_mutex_unlock(mut ));
        _cpu_relax();
        _cpu_relax();
        _cpu_relax();
        pthread_mutex_lock(mut);
        while(*done != 0 && ++w < POLL_REPS) {
                pthread_mutex_unlock(mut);
                _cpu_relax();
                _cpu_relax();
                _cpu_relax();
                pthread_mutex_lock(mut);
        }
        if (*done == 0) {
                ++polls_succ;
                TCHK(pthread_mutex_unlock(mut ));
        } else {
                ++polls_fail;
                TCHK(pthread_mutex_unlock(mut ));
                TCHK(pthread_cond_signal (cond));
        }
}

inline int pthr_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mut, volatile int *valptr)
{
        ERRDECL;
        int rv = 0;
        int w = 0;
        pthread_mutex_lock(mut);
        while (++w < POLL_REP && *valptr == 0) {
                pthread_mutex_unlock(mut);
                _cpu_relax();
                pthread_mutex_lock(mut);
        }
#ifdef DEBUG_THREAD
        fprintf(stderr, "wait, %i iters, val %i\n", w, *valptr);
#endif
        if (*valptr == 0)
                ++poll_fail;
        else
                ++poll_succ;
        while (*valptr == 0)
                rv = pthread_cond_wait(cond, mut);
        *valptr = 0;
        return rv;
}

void lina_err (struct thr_ctrl *tc)
{
        fprintf (stderr, " Thread (%i) synchronization problem!\n", tc->t_no);
        fflush (stderr); abort ();
}

void lina_empty (struct thr_ctrl *dummy)
{
        /* nothing */
}

void* empty_thread (void *dummy)
{
        return NULL;
}

void pthread_mutex_bug_abort (struct thr_ctrl *tc, int err, const char* where)
{
        fprintf (stderr, " Thread %i: cond_wait on %s: %s\n",
                 tc->t_no, where, CSTD__ strerror (err));
        fprintf (stderr, " Most likely caused by pthread bug (introduced by UD after 2.1.3).\n");
        fprintf (stderr, " Get a fixed version or recompile the TBCI SMP support (smp.cc)\n");
        fprintf (stderr, " with HAVE_FASTMUTEX_BUG defined!\n");
        abort ();
}

void* lina_thread (void* thr)
{
        struct thr_ctrl *tc = (struct thr_ctrl*) thr;
        int err;
#if defined(HAVE_TLS) || defined(HAVE_DTLS)
        ismainthread = 0;
        thrno = tc->t_no+1;
        this_thread = tc;
#endif
        tc->t_pid = getpid (); clock (); 
        //tc->t_job = lina_err; tc->t_setup_var = 0;
#ifdef DEBUG_THREAD
        fprintf (stderr, " Thread %i: Try to get done lock\n", tc->t_no);
#endif
        pthread_mutex_lock(&tc->t_done);
        tc->t_res = 0.0; 
        pthr_cond_signal_mutex(&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
#ifdef DEBUG_THREAD
        fprintf (stderr, " Thread %i done with done: id %08lx, pid %i\n",
                 tc->t_no, tc->t_id, tc->t_pid);
#endif
        /* Now wait for some work to do. Finish when a NULL job becomes runnable */
        while (1) {
                /* Set job to error! */
                /* Ready to accept a job: Wait for it */
                /* Implicitly, t_setup mutex is released which allows main thread 
                 * to modify the tc[t_no] struct */
                if (UNLIKELY((err = pthr_cond_wait (&tc->t_setup_cond, &tc->t_setup, &tc->t_setup_done))) != 0)
                        pthread_mutex_bug_abort (tc, err, "setup");
                pthread_mutex_unlock(&tc->t_setup);
#ifdef DEBUG_THREAD
                fprintf (stderr, " Thread %i: Start job %p\n", tc->t_no, tc->t_job); 
                fflush (stderr);
#endif
                if (!tc->t_job) 
                        break; /* The end */
                else 
                        (*tc->t_job) (tc);
                tc->t_job = lina_err;

#ifdef DEBUG_THREAD
                fprintf (stderr, " Thread %i: Job done!\n", tc->t_no); 
                fflush (stderr);
#endif
                /* Wait until main thread waits for us and signal completion */
                pthread_mutex_lock(&tc->t_done);
                pthr_cond_signal_mutex (&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
#ifdef DEBUG_THREAD
                fprintf (stderr, " Thread %i: Signaled Job done!\n", tc->t_no); 
                fflush (stderr);
#endif
        }
#ifdef DEBUG_THREAD
        fprintf (stderr, " Thread %i: exit!\n", tc->t_no);
#endif
        pthread_mutex_lock(&tc->t_done);
        tc->t_retval = clock ();
#if defined(HAVE_CLOCK_GETTIME) && defined(THREAD_STAT)
        struct timespec tm;
        if (!clock_gettime(CLOCK_THREAD_CPUTIME_ID, &tm)) {
                const double secs = tm.tv_sec + (double)tm.tv_nsec/1e9;
                tc->t_retval = (long)(secs * CLOCKS_PER_SEC);
        }
#endif
        pthr_cond_signal_mutex (&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
        pthread_exit (&tc->t_retval);
        return NULL;
}


int init_threads (const int num_cpu)
{
        int t, err, det_cpus;
        pthread_mutexattr_t mutattr;
        
        main_thread_pid = getpid ();
        /* No of CPUs */
        det_cpus  = detect_num_cpu ();
        /* How many ? */
        if (num_cpu <= 0) {
                num_threads = det_cpus;
#if defined(HAVE_GETLOADAVG) && defined(TBCI_SUB_LOAD)
                /* Subtract load */
                int lavg = loadavg ();
                /* If load exceeds no of CPUs, use 2 CPUs */
                if (num_threads - lavg <= 1)
                        num_threads = num_threads > 2? 2: 1;
                else
                        num_threads -= lavg;
#endif
                if (num_cpu && num_threads > -num_cpu)
                        num_threads = -num_cpu;
        } else
                num_threads = num_cpu;
        /* Max is MAX_THREADS */
        if (num_threads > MAX_THREADS) 
                num_threads = MAX_THREADS;
        /* No threads? */
        if (num_threads == 1)
                num_threads = 0;
        /* Sanity check */
        if (num_threads > det_cpus)
                fprintf(stderr, "Warning: Number of threads %i larger than no of CPU cores %i!\n",
                        num_threads, det_cpus);
#ifdef HAVE_SCHED_GETAFFINITY
        else
                sched_setaffinity(main_thread_pid, CPU_SETBYTES, &saved_cpuset);
#endif
        /* Alloc mem for control structs */
        if (num_threads >= 1) 
                threads = (struct thr_ctrl *) memalign(128, sizeof(struct thr_ctrl) * (num_threads));
                //threads = (struct thr_ctrl *) CSTD__ malloc (sizeof(struct thr_ctrl) * (num_threads));
        else 
                threads = NULL;
        if (threads) 
                CSTD__ memset (threads, 0, sizeof(struct thr_ctrl) * (num_threads));
        TCHK(pthread_mutexattr_init (&mutattr));
#ifdef HAVE_FASTMUTEX_BUG
        TCHK(pthread_mutexattr_settype (&mutattr, PTHREAD_MUTEX_ERRORCHECK_NP));
#endif
        /* Some Un*xes only start the clock after the first call to clock() */
        clock ();
#if !defined(HAVE_TLS) && !defined(HAVE_DTLS)
        // If we don't have thread local storage (and no locking), 
        // we need to disable // concurrent memory caching allocations
        // now globally :-(
        if (num_threads >= 1) {
                ismainthread = 0;
                thrno = -1;
        }
#endif
        /* FIXME: Shouldn't we bind threads here already? */
        /* Now init threads */
        for (t = 0; t < num_threads; ++t) {
                struct thr_ctrl *tc = &threads[t];
                //fprintf(stderr, "Debug: thread %i @ %p\n", t, tc);
                tc->t_job = lina_err; tc->t_no = t; 
                //tc->t_done_done = 0; tc->t_setup_done = 0;
                TCHK(pthread_mutex_init (&tc->t_setup, &mutattr));
                TCHK(pthread_cond_init  (&tc->t_setup_cond, NULL));
                TCHK(pthread_mutex_init (&tc->t_done,  &mutattr));
                TCHK(pthread_cond_init  (&tc->t_done_cond, NULL));
                TCHK(pthread_create (&tc->t_id, NULL, lina_thread, tc));
                pthr_cond_wait(&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
                TCHK(pthread_mutex_unlock (&tc->t_done));
        }
#ifdef DEBUG_THREAD
        fprintf (stderr, "%i threads @%p (sz=%zi) set up.\n", 
                        num_threads, threads, sizeof(struct thr_ctrl));
#endif
        TCHK(pthread_mutexattr_destroy (&mutattr));
        for (cback *c = thread_cbacks.getfirst(); c != NULL; c = thread_cbacks.getnext()) {
                //fprintf(stderr, "Callback %p (%p %p %p)\n", c, c->ctor, c->dtor, c->parm);
                c->ctor(c->parm, num_threads);
        }
        //fprintf(stderr, "%li Callbacks registered\n", thread_cbacks.size());

        return num_threads;
}

void bind_threads (bool bind_main)
{
#ifdef HAVE_SCHED_GETAFFINITY
        /* Save CPU affinity mask */
        CPU_ZERO(&saved_cpuset);
        sched_getaffinity (main_thread_pid, CPU_SETBYTES, &saved_cpuset);
        if (num_threads > CPU_COUNT(&saved_cpuset))
                fprintf(stderr, "TBCI SMP: More threads(%i) than allowed CPUs(%i)!\n",
                        num_threads, CPU_COUNT(&saved_cpuset));
        cpu_set_t cpuset; CPU_ZERO(&cpuset);

        int next_cpu = next_set_bit(0, &saved_cpuset);
        if (next_cpu < 0)
                abort();
        bound_main = bind_main;
        if (bind_main) {
                CPU_SET(next_cpu, &cpuset);
                sched_setaffinity(main_thread_pid, CPU_SETBYTES, &cpuset);
#ifdef THREAD_DEBUG             
                fprintf(stderr, "Set Main Thread %i: CPU %i (%s)\n", main_thread_pid,
                        next_cpu, cpu_str(&cpuset));
#endif
#ifdef DEBUG_THREAD
                sched_getaffinity(main_thread_pid, CPU_SETBYTES, &cpuset);
                fprintf(stderr, "Get Main Thread %i: CPU %i (%s)\n", main_thread_pid,
                        next_cpu, cpu_str(&cpuset));
#endif
                CPU_ZERO(&cpuset);
        }
#endif
#ifdef HAVE_PTHREAD_GETAFFINITY_NP
        threads_bound = true;
        for (int t = 0; t < num_threads; ++t) {
                struct thr_ctrl *tc = &threads[t];
#if 0
                // Special case: Half as many threads as cores
                if (num_threads == CPU_COUNT(&saved_cpuset)/2)
                        ++next_cpu;
#endif
                next_cpu = next_set_bit(next_cpu+1, &saved_cpuset);
                if (next_cpu == -1) 
                        next_cpu = next_set_bit(next_cpu+1, &saved_cpuset);
                CPU_SET(next_cpu, &cpuset);
                pthread_setaffinity_np(tc->t_id, CPU_SETBYTES, &cpuset);
#ifdef THREAD_DEBUG
                fprintf(stderr, "Set Thread %i: CPUs %s\n", t, 
                        cpu_str(&cpuset));
#endif
#ifdef DEBUG_THREAD
                pthread_getaffinity_np(tc->t_id, CPU_SETBYTES, &cpuset);
                fprintf(stderr, "Get Thread %i: CPUs %s\n", t, 
                        cpu_str(&cpuset));
#endif
                CPU_ZERO(&cpuset);
        }
#endif
}

int tot_cpu_tm;
void free_threads ()
{
        int t;
        tot_cpu_tm = 0;
        ERRDECL;
        //fprintf (stderr, "Free threads ...\n");
        for (t = 0; t < num_threads; ++t) {
                struct thr_ctrl *tc = &threads[t];
                void *res = &tc->t_retval;
                /* Make sure, the worker is waiting for a job */
                TCHK(pthread_mutex_lock (&tc->t_setup));
                tc->t_job = 0; 
                pthr_cond_signal_mutex(&tc->t_setup_cond, &tc->t_setup, &tc->t_setup_done);
                TCHK(pthread_join (tc->t_id, &res));
                TCHK(pthread_mutex_unlock (&tc->t_done));
                TCHK(pthread_mutex_destroy (&tc->t_setup));
                TCHK(pthread_cond_destroy (&tc->t_setup_cond));
                TCHK(pthread_mutex_destroy (&tc->t_done));
                TCHK(pthread_cond_destroy (&tc->t_done_cond));
                long tm = *(long*)res;
                tot_cpu_tm += tm;
#ifdef THREAD_STAT
                fprintf (stderr, " CPU time for thread %i  :%7.3f s\n", 
                         tc->t_no, (double)(tm)/CLOCKS_PER_SEC);
#endif
        }
        for (cback *c = thread_cbacks.getfirst(); c != NULL; c = thread_cbacks.getnext()) {
                //fprintf(stderr, "Callback %p (%p %p %p)\n", c, c->ctor, c->dtor, c->parm);
                c->dtor(c->parm, num_threads);
        }
        clock_t mainclock = clock();
#ifdef HAVE_CLOCK_GETTIME
        struct timespec tm;
        if (!clock_gettime(CLOCK_THREAD_CPUTIME_ID, &tm)) {
                double secs = tm.tv_sec + (double)tm.tv_nsec/1e9;
                mainclock = (long)(secs*CLOCKS_PER_SEC);
        }
#endif
        tot_cpu_tm += mainclock;
#ifdef THREAD_STAT
        fprintf (stderr, " CPU time for main thr  :%7.3f s\n", 
                 (double)mainclock/CLOCKS_PER_SEC);
        fprintf (stderr, " CPU for all threads    :%7.3f s\n",
                 (double)(tot_cpu_tm)/CLOCKS_PER_SEC);
        fprintf (stderr, " Wait Poll success: %li failures: %li\n", poll_succ, poll_fail);
        fprintf (stderr, " Sign Poll success: %li failures: %li\n", polls_succ, polls_fail);
        fflush (stderr);
#endif
        if (num_threads > 0) 
                CSTD__ free (threads); 
        num_threads = 0; threads = 0;
#if !defined(HAVE_TLS) && !defined(HAVE_DTLS)
        ismainthread = 1;
        thrno = 0;
#endif
#ifdef HAVE_SCHED_GETAFFINITY
        if (bound_main) {
                sched_setaffinity(main_thread_pid, CPU_SETBYTES, &saved_cpuset);
                bound_main = false;
        }
        threads_bound = false;
#endif
}

void _thread_start_off (const int thr_no, thr_job_t job, 
                        const unsigned long off, const unsigned long sz,
                        va_list vl)
{
        void * par; unsigned t = 0;
        struct thr_ctrl* tc = &threads[thr_no];
        ERRDECL;
        //va_list (vl);
        BCHKNR(thr_no >= num_threads, NumErr, Starting thread no outside range, thr_no);
        threads_busy++;
        TCHK(pthread_mutex_lock (&tc->t_setup));
        tc->t_size = sz; tc->t_off = off;
        tc->t_job = job; 
        while ((par = va_arg (vl, void*))) 
                tc->t_par[t++] = par;
        BCHKNR(t > THREAD_MAX_ARGS, NumErr, Too many arguments to thread_start, t-1);
        /* And now: the thread is supposed to do some work */
        pthr_cond_signal_mutex(&tc->t_setup_cond, &tc->t_setup, &tc->t_setup_done);
#ifdef DEBUG_THREAD
        fprintf (stderr, "Setup lock %i released and signal sent\n", tc->t_no);
#endif
}

void thread_start_off (const int thr_no, thr_job_t job, 
                       const unsigned long off, const unsigned long sz, ...)
{
        va_list(vl);
        va_start(vl, sz);
        _thread_start_off(thr_no, job, off, sz, vl);
        va_end(vl);
}

void thread_start ( const int thr_no, thr_job_t job, 
                    const unsigned long sz, ...)
{
        va_list(vl);
        va_start(vl, sz);
        _thread_start_off(thr_no, job, 0, sz, vl);
        va_end(vl);
}

void thread_wait (const int thr_no)
{
        struct thr_ctrl *tc = &threads[thr_no];
        BCHKNR (thr_no >= num_threads, NumErr, Wait for non-existing thread, thr_no);
        /* Tell the thread that we wait for completion */
#ifdef DEBUG_THREAD
        fprintf (stderr, "Wait for thread %i\n", tc->t_no);
#endif
        const int err = pthr_cond_wait (&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
        if (UNLIKELY(err))
                pthread_mutex_bug_abort (tc, err, "done");
        threads_busy--;
        pthread_mutex_unlock(&tc->t_done);
#ifdef DEBUG_THREAD
        fprintf (stderr, "Thread %i signaled completion.\n", tc->t_no);
#endif
}

void* thread_wait_useful (const int thr_no, useful_job_t job, void* arg)
{
        struct thr_ctrl *tc = &threads[thr_no];
        void* rv = NULL;
        ERRDECL;
        TCHK(pthread_mutex_lock (&tc->t_done));
        while (tc->t_done_done == 0) {
                pthread_mutex_unlock(&tc->t_done);
                if (job) 
                        rv = (*job)(arg);
                else
                        _cpu_relax();
                pthread_mutex_lock (&tc->t_done);
        }
        TCHK(pthread_mutex_unlock (&tc->t_done));
        return rv;
}

LONG_DOUBLE thread_wait_result (const int thr_no)
{
        struct thr_ctrl *tc = &threads[thr_no];
        int err;
        BCHKNR (thr_no >= num_threads, NumErr, Wait for non-existing thread, thr_no);
#ifdef DEBUG_THREAD
        fprintf (stderr, "Wait for result of thread %i\n", tc->t_no);
#endif
        err = pthr_cond_wait (&tc->t_done_cond, &tc->t_done, &tc->t_done_done);
        if (err) 
                pthread_mutex_bug_abort (tc, err, "done");
        threads_busy--;
        LONG_DOUBLE res = tc->t_res;
        TCHK(pthread_mutex_unlock (&tc->t_done));
#ifdef DEBUG_THREAD
        fprintf (stderr, "Thread %i signaled completion\n", tc->t_no);
#endif
        return /*(volatile LONG_DOUBLE )*/res;
}

int threads_avail (int wanted)
{
        if (threads_busy) {
#ifdef THREAD_DEBUG
                //fprintf (stderr, "Threads busy!\n"); fflush (stderr);
#endif
                return 0; /* avoid deadlock */
        }
        if (UNLIKELY(wanted > num_threads)) 
                return num_threads;
        else 
                return wanted;
}

void disable_threads()
{
        threads_busy++;
}

void reenable_threads ()
{
        if (threads_busy)
                threads_busy--;
        else
                fprintf (stderr, "reenable_threads(): Threads already enabled.!\n");
}

NAMESPACE_END

#else /* no SMP */
/* These are for compatibility ... */

NAMESPACE_TBCI
pid_t main_thread_pid WEAKA = 0;

WEAK(int  init_threads (const int c))
{ main_thread_pid = getpid(); clock (); return 0; }
WEAK(void free_threads ()) {}
WEAK(int  threads_avail (const int w)) { return 0; }
WEAK(void disable_threads ()) {}
WEAK(void reenable_threads ()) {}
WEAK(void thread_start (const int tno, thr_job_t job,
                        const unsigned long sz, ...))
{
        void * par; 
        struct thr_ctrl thrc; struct thr_ctrl* tc = &thrc;
        unsigned int t = 0;
        va_list (vl);
        tc->t_job = job; tc->t_size = sz; tc->t_off = 0;
        va_start (vl, sz);
        while ((par = va_arg (vl, void*)))
                tc->t_par[t++] = par;
        va_end (vl);
        fprintf (stderr, "Warning: Tried to start a thread with non-SMP compiled smp.cc !\n");
# ifdef ABORT_ON_ERR
        abort ();
# endif
        (*job)(tc);
}
WEAK(void thread_start_off (const int tno, thr_job_t job,
                            const unsigned long off, const unsigned long sz, ...))
{
        void * par;
        struct thr_ctrl thrc; struct thr_ctrl* tc = &thrc;
        unsigned int t = 0;
        va_list (vl);
        tc->t_job = job; tc->t_size = sz; tc->t_off = off;
        va_start (vl, sz);
        while ((par = va_arg (vl, void*)))
                tc->t_par[t++] = par;
        va_end (vl);
        fprintf (stderr, "Warning: Tried to start a thread with non-SMP compiled smp.cc !\n");
# ifdef ABORT_ON_ERR
        abort ();
#endif
        (*job)(tc);
}
WEAK(void  thread_wait (const int t)) {}
WEAK(void* thread_wait_useful (const int t, useful_job_t j, void* a))
{ return 0; }
WEAK(LONG_DOUBLE thread_wait_result (const int thr_no)) 
{ return 0; }


NAMESPACE_END

#endif /* SMP */
#endif /* _POSIX_THREADS */

---