/* * functions about threads. * * Copyright (C) 2017 Christopher Fauet - cfaulet@haproxy.com * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public License * as published by the Free Software Foundation; either version * 2 of the License, or (at your option) any later version. * */ #define _GNU_SOURCE #include #include #include #ifdef USE_CPU_AFFINITY #include #endif #ifdef __FreeBSD__ #include #endif #include #include #include #include #include struct thread_info ha_thread_info[MAX_THREADS] = { }; THREAD_LOCAL struct thread_info *ti = &ha_thread_info[0]; #ifdef USE_THREAD volatile unsigned long threads_want_rdv_mask = 0; volatile unsigned long threads_harmless_mask = 0; volatile unsigned long threads_sync_mask = 0; volatile unsigned long all_threads_mask = 1; // nbthread 1 assumed by default THREAD_LOCAL unsigned int tid = 0; THREAD_LOCAL unsigned long tid_bit = (1UL << 0); int thread_cpus_enabled_at_boot = 1; #if defined(DEBUG_THREAD) || defined(DEBUG_FULL) struct lock_stat lock_stats[LOCK_LABELS]; #endif /* Marks the thread as harmless until the last thread using the rendez-vous * point quits. Given that we can wait for a long time, sched_yield() is used * when available to offer the CPU resources to competing threads if needed. */ void thread_harmless_till_end() { _HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); while (threads_want_rdv_mask & all_threads_mask) { ha_thread_relax(); } } /* Isolates the current thread : request the ability to work while all other * threads are harmless. Only returns once all of them are harmless, with the * current thread's bit in threads_harmless_mask cleared. Needs to be completed * using thread_release(). */ void thread_isolate() { unsigned long old; _HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); __ha_barrier_atomic_store(); _HA_ATOMIC_OR(&threads_want_rdv_mask, tid_bit); /* wait for all threads to become harmless */ old = threads_harmless_mask; while (1) { if (unlikely((old & all_threads_mask) != all_threads_mask)) old = threads_harmless_mask; else if (_HA_ATOMIC_CAS(&threads_harmless_mask, &old, old & ~tid_bit)) break; ha_thread_relax(); } /* one thread gets released at a time here, with its harmess bit off. * The loss of this bit makes the other one continue to spin while the * thread is working alone. */ } /* Cancels the effect of thread_isolate() by releasing the current thread's bit * in threads_want_rdv_mask and by marking this thread as harmless until the * last worker finishes. */ void thread_release() { _HA_ATOMIC_AND(&threads_want_rdv_mask, ~tid_bit); while (threads_want_rdv_mask & all_threads_mask) { _HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); while (threads_want_rdv_mask & all_threads_mask) ha_thread_relax(); HA_ATOMIC_AND(&threads_harmless_mask, ~tid_bit); } } /* Cancels the effect of thread_isolate() by releasing the current thread's bit * in threads_want_rdv_mask and by marking this thread as harmless until the * last worker finishes. The difference with thread_release() is that this one * will not leave the function before others are notified to do the same, so it * guarantees that the current thread will not pass through a subsequent call * to thread_isolate() before others finish. */ void thread_sync_release() { _HA_ATOMIC_OR(&threads_sync_mask, tid_bit); __ha_barrier_atomic_store(); _HA_ATOMIC_AND(&threads_want_rdv_mask, ~tid_bit); while (threads_want_rdv_mask & all_threads_mask) { _HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); while (threads_want_rdv_mask & all_threads_mask) ha_thread_relax(); HA_ATOMIC_AND(&threads_harmless_mask, ~tid_bit); } /* the current thread is not harmless anymore, thread_isolate() * is forced to wait till all waiters finish. */ _HA_ATOMIC_AND(&threads_sync_mask, ~tid_bit); while (threads_sync_mask & all_threads_mask) ha_thread_relax(); } /* send signal to thread */ void ha_tkill(unsigned int thr, int sig) { pthread_kill(ha_thread_info[thr].pthread, sig); } /* send signal to all threads. The calling thread is signaled last in * order to allow all threads to synchronize in the handler. */ void ha_tkillall(int sig) { unsigned int thr; for (thr = 0; thr < global.nbthread; thr++) { if (!(all_threads_mask & (1UL << thr))) continue; if (thr == tid) continue; pthread_kill(ha_thread_info[thr].pthread, sig); } raise(sig); } /* these calls are used as callbacks at init time */ void ha_spin_init(HA_SPINLOCK_T *l) { HA_SPIN_INIT(l); } /* these calls are used as callbacks at init time */ void ha_rwlock_init(HA_RWLOCK_T *l) { HA_RWLOCK_INIT(l); } /* returns the number of CPUs the current process is enabled to run on */ static int thread_cpus_enabled() { int ret = 1; #ifdef USE_CPU_AFFINITY #if defined(__linux__) && defined(CPU_COUNT) cpu_set_t mask; if (sched_getaffinity(0, sizeof(mask), &mask) == 0) ret = CPU_COUNT(&mask); #elif defined(__FreeBSD__) && defined(USE_CPU_AFFINITY) cpuset_t cpuset; if (cpuset_getaffinity(CPU_LEVEL_CPUSET, CPU_WHICH_PID, -1, sizeof(cpuset), &cpuset) == 0) ret = CPU_COUNT(&cpuset); #endif #endif ret = MAX(ret, 1); ret = MIN(ret, MAX_THREADS); return ret; } /* Depending on the platform and how libpthread was built, pthread_exit() may * involve some code in libgcc_s that would be loaded on exit for the first * time, causing aborts if the process is chrooted. It's harmless bit very * dirty. There isn't much we can do to make sure libgcc_s is loaded only if * needed, so what we do here is that during early boot we create a dummy * thread that immediately exits. This will lead to libgcc_s being loaded * during boot on the platforms where it's required. */ static void *dummy_thread_function(void *data) { pthread_exit(NULL); return NULL; } static inline void preload_libgcc_s(void) { pthread_t dummy_thread; pthread_create(&dummy_thread, NULL, dummy_thread_function, NULL); pthread_join(dummy_thread, NULL); } __attribute__((constructor)) static void __thread_init(void) { char *ptr = NULL; if (MAX_THREADS < 1 || MAX_THREADS > LONGBITS) { ha_alert("MAX_THREADS value must be between 1 and %d inclusive; " "HAProxy was built with value %d, please fix it and rebuild.\n", LONGBITS, MAX_THREADS); exit(1); } preload_libgcc_s(); thread_cpus_enabled_at_boot = thread_cpus_enabled(); memprintf(&ptr, "Built with multi-threading support (MAX_THREADS=%d, default=%d).", MAX_THREADS, thread_cpus_enabled_at_boot); hap_register_build_opts(ptr, 1); #if defined(DEBUG_THREAD) || defined(DEBUG_FULL) memset(lock_stats, 0, sizeof(lock_stats)); #endif } #else REGISTER_BUILD_OPTS("Built without multi-threading support (USE_THREAD not set)."); #endif // USE_THREAD /* Parse the number of threads in argument , returns it and adjusts a few * internal variables accordingly, or fails and returns zero with an error * reason in . May be called multiple times while parsing. */ int parse_nbthread(const char *arg, char **err) { long nbthread; char *errptr; nbthread = strtol(arg, &errptr, 10); if (!*arg || *errptr) { memprintf(err, "passed a missing or unparsable integer value in '%s'", arg); return 0; } #ifndef USE_THREAD if (nbthread != 1) { memprintf(err, "specified with a value other than 1 while HAProxy is not compiled with threads support. Please check build options for USE_THREAD"); return 0; } #else if (nbthread < 1 || nbthread > MAX_THREADS) { memprintf(err, "value must be between 1 and %d (was %ld)", MAX_THREADS, nbthread); return 0; } all_threads_mask = nbits(nbthread); #endif return nbthread; }