summaryrefslogtreecommitdiffstats
path: root/src/thread.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/thread.c1864
1 files changed, 1864 insertions, 0 deletions
diff --git a/src/thread.c b/src/thread.c
new file mode 100644
index 0000000..ab4342d
--- /dev/null
+++ b/src/thread.c
@@ -0,0 +1,1864 @@
+/*
+ * 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 <unistd.h>
+#include <stdlib.h>
+
+#include <signal.h>
+#include <unistd.h>
+#ifdef _POSIX_PRIORITY_SCHEDULING
+#include <sched.h>
+#endif
+
+#ifdef USE_THREAD
+# include <pthread.h>
+#endif
+
+#ifdef USE_CPU_AFFINITY
+# include <sched.h>
+# if defined(__FreeBSD__) || defined(__DragonFly__)
+# include <sys/param.h>
+# ifdef __FreeBSD__
+# include <sys/cpuset.h>
+# endif
+# include <pthread_np.h>
+# endif
+# ifdef __APPLE__
+# include <mach/mach_types.h>
+# include <mach/thread_act.h>
+# include <mach/thread_policy.h>
+# endif
+# include <haproxy/cpuset.h>
+#endif
+
+#include <haproxy/cfgparse.h>
+#include <haproxy/clock.h>
+#include <haproxy/fd.h>
+#include <haproxy/global.h>
+#include <haproxy/log.h>
+#include <haproxy/thread.h>
+#include <haproxy/tools.h>
+
+struct tgroup_info ha_tgroup_info[MAX_TGROUPS] = { };
+THREAD_LOCAL const struct tgroup_info *tg = &ha_tgroup_info[0];
+
+struct thread_info ha_thread_info[MAX_THREADS] = { };
+THREAD_LOCAL const struct thread_info *ti = &ha_thread_info[0];
+
+struct tgroup_ctx ha_tgroup_ctx[MAX_TGROUPS] = { };
+THREAD_LOCAL struct tgroup_ctx *tg_ctx = &ha_tgroup_ctx[0];
+
+struct thread_ctx ha_thread_ctx[MAX_THREADS] = { };
+THREAD_LOCAL struct thread_ctx *th_ctx = &ha_thread_ctx[0];
+
+#ifdef USE_THREAD
+
+volatile unsigned long all_tgroups_mask __read_mostly = 1; // nbtgroup 1 assumed by default
+volatile unsigned int rdv_requests = 0; // total number of threads requesting RDV
+volatile unsigned int isolated_thread = ~0; // ID of the isolated thread, or ~0 when none
+THREAD_LOCAL unsigned int tgid = 1; // thread ID starts at 1
+THREAD_LOCAL unsigned int tid = 0;
+int thread_cpus_enabled_at_boot = 1;
+static pthread_t ha_pthread[MAX_THREADS] = { };
+
+/* 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(&tg_ctx->threads_harmless, ti->ltid_bit);
+ while (_HA_ATOMIC_LOAD(&rdv_requests) != 0) {
+ ha_thread_relax();
+ }
+}
+
+/* Isolates the current thread : request the ability to work while all other
+ * threads are harmless, as defined by thread_harmless_now() (i.e. they're not
+ * going to touch any visible memory area). Only returns once all of them are
+ * harmless, with the current thread's bit in &tg_ctx->threads_harmless cleared.
+ * Needs to be completed using thread_release().
+ */
+void thread_isolate()
+{
+ uint tgrp, thr;
+
+ _HA_ATOMIC_OR(&tg_ctx->threads_harmless, ti->ltid_bit);
+ __ha_barrier_atomic_store();
+ _HA_ATOMIC_INC(&rdv_requests);
+
+ /* wait for all threads to become harmless. They cannot change their
+ * mind once seen thanks to rdv_requests above, unless they pass in
+ * front of us. For this reason we proceed in 4 steps:
+ * 1) wait for all threads to declare themselves harmless
+ * 2) try to grab the isolated_thread exclusivity
+ * 3) verify again that all threads are harmless, since another one
+ * that was isolating between 1 and 2 could have dropped its
+ * harmless state there.
+ * 4) drop harmless flag (which also has the benefit of leaving
+ * all other threads wait on reads instead of writes.
+ */
+ while (1) {
+ for (tgrp = 0; tgrp < global.nbtgroups; tgrp++) {
+ do {
+ ulong te = _HA_ATOMIC_LOAD(&ha_tgroup_info[tgrp].threads_enabled);
+ ulong th = _HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_harmless);
+
+ if ((th & te) == te)
+ break;
+ ha_thread_relax();
+ } while (1);
+ }
+
+ /* all other ones are harmless. isolated_thread will contain
+ * ~0U if no other one competes, !=tid if another one got it,
+ * tid if the current thread already grabbed it on the previous
+ * round.
+ */
+ thr = _HA_ATOMIC_LOAD(&isolated_thread);
+ if (thr == tid)
+ break; // we won and we're certain everyone is harmless
+
+ /* try to win the race against others */
+ if (thr != ~0U || !_HA_ATOMIC_CAS(&isolated_thread, &thr, tid))
+ ha_thread_relax();
+ }
+
+ /* the thread is no longer harmless as it runs */
+ _HA_ATOMIC_AND(&tg_ctx->threads_harmless, ~ti->ltid_bit);
+
+ /* the thread is isolated until it calls thread_release() which will
+ * 1) reset isolated_thread to ~0;
+ * 2) decrement rdv_requests.
+ */
+}
+
+/* Isolates the current thread : request the ability to work while all other
+ * threads are idle, as defined by thread_idle_now(). It only returns once
+ * all of them are both harmless and idle, with the current thread's bit in
+ * &tg_ctx->threads_harmless and idle_mask cleared. Needs to be completed using
+ * thread_release(). By doing so the thread also engages in being safe against
+ * any actions that other threads might be about to start under the same
+ * conditions. This specifically targets destruction of any internal structure,
+ * which implies that the current thread may not hold references to any object.
+ *
+ * Note that a concurrent thread_isolate() will usually win against
+ * thread_isolate_full() as it doesn't consider the idle_mask, allowing it to
+ * get back to the poller or any other fully idle location, that will
+ * ultimately release this one.
+ */
+void thread_isolate_full()
+{
+ uint tgrp, thr;
+
+ _HA_ATOMIC_OR(&tg_ctx->threads_idle, ti->ltid_bit);
+ _HA_ATOMIC_OR(&tg_ctx->threads_harmless, ti->ltid_bit);
+ __ha_barrier_atomic_store();
+ _HA_ATOMIC_INC(&rdv_requests);
+
+ /* wait for all threads to become harmless. They cannot change their
+ * mind once seen thanks to rdv_requests above, unless they pass in
+ * front of us. For this reason we proceed in 4 steps:
+ * 1) wait for all threads to declare themselves harmless
+ * 2) try to grab the isolated_thread exclusivity
+ * 3) verify again that all threads are harmless, since another one
+ * that was isolating between 1 and 2 could have dropped its
+ * harmless state there.
+ * 4) drop harmless flag (which also has the benefit of leaving
+ * all other threads wait on reads instead of writes.
+ */
+ while (1) {
+ for (tgrp = 0; tgrp < global.nbtgroups; tgrp++) {
+ do {
+ ulong te = _HA_ATOMIC_LOAD(&ha_tgroup_info[tgrp].threads_enabled);
+ ulong th = _HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_harmless);
+ ulong id = _HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_idle);
+
+ if ((th & id & te) == te)
+ break;
+ ha_thread_relax();
+ } while (1);
+ }
+
+ /* all other ones are harmless and idle. isolated_thread will
+ * contain ~0U if no other one competes, !=tid if another one
+ * got it, tid if the current thread already grabbed it on the
+ * previous round.
+ */
+ thr = _HA_ATOMIC_LOAD(&isolated_thread);
+ if (thr == tid)
+ break; // we won and we're certain everyone is harmless
+
+ if (thr != ~0U || !_HA_ATOMIC_CAS(&isolated_thread, &thr, tid))
+ ha_thread_relax();
+ }
+
+ /* we're not idle nor harmless anymore at this point. Other threads
+ * waiting on this condition will need to wait until out next pass to
+ * the poller, or our next call to thread_isolate_full().
+ */
+ _HA_ATOMIC_AND(&tg_ctx->threads_idle, ~ti->ltid_bit);
+ _HA_ATOMIC_AND(&tg_ctx->threads_harmless, ~ti->ltid_bit);
+
+ /* the thread is isolated until it calls thread_release() which will
+ * 1) reset isolated_thread to ~0;
+ * 2) decrement rdv_requests.
+ */
+}
+
+/* Cancels the effect of thread_isolate() by resetting the ID of the isolated
+ * thread and decrementing the number of RDV requesters. This immediately allows
+ * other threads to expect to be executed, though they will first have to wait
+ * for this thread to become harmless again (possibly by reaching the poller
+ * again).
+ */
+void thread_release()
+{
+ HA_ATOMIC_STORE(&isolated_thread, ~0U);
+ HA_ATOMIC_DEC(&rdv_requests);
+}
+
+/* Sets up threads, signals and masks, and starts threads 2 and above.
+ * Does nothing when threads are disabled.
+ */
+void setup_extra_threads(void *(*handler)(void *))
+{
+ sigset_t blocked_sig, old_sig;
+ int i;
+
+ /* ensure the signals will be blocked in every thread */
+ sigfillset(&blocked_sig);
+ sigdelset(&blocked_sig, SIGPROF);
+ sigdelset(&blocked_sig, SIGBUS);
+ sigdelset(&blocked_sig, SIGFPE);
+ sigdelset(&blocked_sig, SIGILL);
+ sigdelset(&blocked_sig, SIGSEGV);
+ pthread_sigmask(SIG_SETMASK, &blocked_sig, &old_sig);
+
+ /* Create nbthread-1 thread. The first thread is the current process */
+ ha_pthread[0] = pthread_self();
+ for (i = 1; i < global.nbthread; i++)
+ pthread_create(&ha_pthread[i], NULL, handler, &ha_thread_info[i]);
+}
+
+/* waits for all threads to terminate. Does nothing when threads are
+ * disabled.
+ */
+void wait_for_threads_completion()
+{
+ int i;
+
+ /* Wait the end of other threads */
+ for (i = 1; i < global.nbthread; i++)
+ pthread_join(ha_pthread[i], NULL);
+
+#if defined(DEBUG_THREAD) || defined(DEBUG_FULL)
+ show_lock_stats();
+#endif
+}
+
+/* Tries to set the current thread's CPU affinity according to the cpu_map */
+void set_thread_cpu_affinity()
+{
+#if defined(USE_CPU_AFFINITY)
+ /* no affinity setting for the master process */
+ if (master)
+ return;
+
+ /* Now the CPU affinity for all threads */
+ if (ha_cpuset_count(&cpu_map[tgid - 1].thread[ti->ltid])) {/* only do this if the thread has a THREAD map */
+# if defined(__APPLE__)
+ /* Note: this API is limited to the first 32/64 CPUs */
+ unsigned long set = cpu_map[tgid - 1].thread[ti->ltid].cpuset;
+ int j;
+
+ while ((j = ffsl(set)) > 0) {
+ thread_affinity_policy_data_t cpu_set = { j - 1 };
+ thread_port_t mthread;
+
+ mthread = pthread_mach_thread_np(ha_pthread[tid]);
+ thread_policy_set(mthread, THREAD_AFFINITY_POLICY, (thread_policy_t)&cpu_set, 1);
+ set &= ~(1UL << (j - 1));
+ }
+# else
+ struct hap_cpuset *set = &cpu_map[tgid - 1].thread[ti->ltid];
+
+ pthread_setaffinity_np(ha_pthread[tid], sizeof(set->cpuset), &set->cpuset);
+# endif
+ }
+#endif /* USE_CPU_AFFINITY */
+}
+
+/* Retrieves the opaque pthread_t of thread <thr> cast to an unsigned long long
+ * since POSIX took great care of not specifying its representation, making it
+ * hard to export for post-mortem analysis. For this reason we copy it into a
+ * union and will use the smallest scalar type at least as large as its size,
+ * which will keep endianness and alignment for all regular sizes. As a last
+ * resort we end up with a long long ligned to the first bytes in memory, which
+ * will be endian-dependent if pthread_t is larger than a long long (not seen
+ * yet).
+ */
+unsigned long long ha_get_pthread_id(unsigned int thr)
+{
+ union {
+ pthread_t t;
+ unsigned long long ll;
+ unsigned int i;
+ unsigned short s;
+ unsigned char c;
+ } u = { 0 };
+
+ u.t = ha_pthread[thr];
+
+ if (sizeof(u.t) <= sizeof(u.c))
+ return u.c;
+ else if (sizeof(u.t) <= sizeof(u.s))
+ return u.s;
+ else if (sizeof(u.t) <= sizeof(u.i))
+ return u.i;
+ return u.ll;
+}
+
+/* send signal <sig> to thread <thr> */
+void ha_tkill(unsigned int thr, int sig)
+{
+ pthread_kill(ha_pthread[thr], sig);
+}
+
+/* send signal <sig> 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 (!(ha_thread_info[thr].tg->threads_enabled & ha_thread_info[thr].ltid_bit))
+ continue;
+ if (thr == tid)
+ continue;
+ pthread_kill(ha_pthread[thr], sig);
+ }
+ raise(sig);
+}
+
+void ha_thread_relax(void)
+{
+#ifdef _POSIX_PRIORITY_SCHEDULING
+ sched_yield();
+#else
+ pl_cpu_relax();
+#endif
+}
+
+/* these calls are used as callbacks at init time when debugging is on */
+void ha_spin_init(HA_SPINLOCK_T *l)
+{
+ HA_SPIN_INIT(l);
+}
+
+/* these calls are used as callbacks at init time when debugging is on */
+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,
+ * regardless of any MAX_THREADS limitation.
+ */
+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);
+#elif defined(__APPLE__)
+ ret = (int)sysconf(_SC_NPROCESSORS_ONLN);
+#endif
+#endif
+ ret = MAX(ret, 1);
+ return ret;
+}
+
+/* Returns 1 if the cpu set is currently restricted for the process else 0.
+ * Currently only implemented for the Linux platform.
+ */
+int thread_cpu_mask_forced()
+{
+#if defined(__linux__)
+ const int cpus_avail = sysconf(_SC_NPROCESSORS_ONLN);
+ return cpus_avail != thread_cpus_enabled();
+#else
+ return 0;
+#endif
+}
+
+/* Below come the lock-debugging functions */
+
+#if defined(DEBUG_THREAD) || defined(DEBUG_FULL)
+
+struct lock_stat lock_stats[LOCK_LABELS];
+
+/* this is only used below */
+static const char *lock_label(enum lock_label label)
+{
+ switch (label) {
+ case TASK_RQ_LOCK: return "TASK_RQ";
+ case TASK_WQ_LOCK: return "TASK_WQ";
+ case LISTENER_LOCK: return "LISTENER";
+ case PROXY_LOCK: return "PROXY";
+ case SERVER_LOCK: return "SERVER";
+ case LBPRM_LOCK: return "LBPRM";
+ case SIGNALS_LOCK: return "SIGNALS";
+ case STK_TABLE_LOCK: return "STK_TABLE";
+ case STK_SESS_LOCK: return "STK_SESS";
+ case APPLETS_LOCK: return "APPLETS";
+ case PEER_LOCK: return "PEER";
+ case SHCTX_LOCK: return "SHCTX";
+ case SSL_LOCK: return "SSL";
+ case SSL_GEN_CERTS_LOCK: return "SSL_GEN_CERTS";
+ case PATREF_LOCK: return "PATREF";
+ case PATEXP_LOCK: return "PATEXP";
+ case VARS_LOCK: return "VARS";
+ case COMP_POOL_LOCK: return "COMP_POOL";
+ case LUA_LOCK: return "LUA";
+ case NOTIF_LOCK: return "NOTIF";
+ case SPOE_APPLET_LOCK: return "SPOE_APPLET";
+ case DNS_LOCK: return "DNS";
+ case PID_LIST_LOCK: return "PID_LIST";
+ case EMAIL_ALERTS_LOCK: return "EMAIL_ALERTS";
+ case PIPES_LOCK: return "PIPES";
+ case TLSKEYS_REF_LOCK: return "TLSKEYS_REF";
+ case AUTH_LOCK: return "AUTH";
+ case RING_LOCK: return "RING";
+ case DICT_LOCK: return "DICT";
+ case PROTO_LOCK: return "PROTO";
+ case QUEUE_LOCK: return "QUEUE";
+ case CKCH_LOCK: return "CKCH";
+ case SNI_LOCK: return "SNI";
+ case SSL_SERVER_LOCK: return "SSL_SERVER";
+ case SFT_LOCK: return "SFT";
+ case IDLE_CONNS_LOCK: return "IDLE_CONNS";
+ case OCSP_LOCK: return "OCSP";
+ case QC_CID_LOCK: return "QC_CID";
+ case CACHE_LOCK: return "CACHE";
+ case OTHER_LOCK: return "OTHER";
+ case DEBUG1_LOCK: return "DEBUG1";
+ case DEBUG2_LOCK: return "DEBUG2";
+ case DEBUG3_LOCK: return "DEBUG3";
+ case DEBUG4_LOCK: return "DEBUG4";
+ case DEBUG5_LOCK: return "DEBUG5";
+ case LOCK_LABELS: break; /* keep compiler happy */
+ };
+ /* only way to come here is consecutive to an internal bug */
+ abort();
+}
+
+void show_lock_stats()
+{
+ int lbl;
+
+ for (lbl = 0; lbl < LOCK_LABELS; lbl++) {
+ if (!lock_stats[lbl].num_write_locked &&
+ !lock_stats[lbl].num_seek_locked &&
+ !lock_stats[lbl].num_read_locked) {
+ fprintf(stderr,
+ "Stats about Lock %s: not used\n",
+ lock_label(lbl));
+ continue;
+ }
+
+ fprintf(stderr,
+ "Stats about Lock %s: \n",
+ lock_label(lbl));
+
+ if (lock_stats[lbl].num_write_locked)
+ fprintf(stderr,
+ "\t # write lock : %llu\n"
+ "\t # write unlock: %llu (%lld)\n"
+ "\t # wait time for write : %.3f msec\n"
+ "\t # wait time for write/lock: %.3f nsec\n",
+ (ullong)lock_stats[lbl].num_write_locked,
+ (ullong)lock_stats[lbl].num_write_unlocked,
+ (llong)(lock_stats[lbl].num_write_unlocked - lock_stats[lbl].num_write_locked),
+ (double)lock_stats[lbl].nsec_wait_for_write / 1000000.0,
+ lock_stats[lbl].num_write_locked ? ((double)lock_stats[lbl].nsec_wait_for_write / (double)lock_stats[lbl].num_write_locked) : 0);
+
+ if (lock_stats[lbl].num_seek_locked)
+ fprintf(stderr,
+ "\t # seek lock : %llu\n"
+ "\t # seek unlock : %llu (%lld)\n"
+ "\t # wait time for seek : %.3f msec\n"
+ "\t # wait time for seek/lock : %.3f nsec\n",
+ (ullong)lock_stats[lbl].num_seek_locked,
+ (ullong)lock_stats[lbl].num_seek_unlocked,
+ (llong)(lock_stats[lbl].num_seek_unlocked - lock_stats[lbl].num_seek_locked),
+ (double)lock_stats[lbl].nsec_wait_for_seek / 1000000.0,
+ lock_stats[lbl].num_seek_locked ? ((double)lock_stats[lbl].nsec_wait_for_seek / (double)lock_stats[lbl].num_seek_locked) : 0);
+
+ if (lock_stats[lbl].num_read_locked)
+ fprintf(stderr,
+ "\t # read lock : %llu\n"
+ "\t # read unlock : %llu (%lld)\n"
+ "\t # wait time for read : %.3f msec\n"
+ "\t # wait time for read/lock : %.3f nsec\n",
+ (ullong)lock_stats[lbl].num_read_locked,
+ (ullong)lock_stats[lbl].num_read_unlocked,
+ (llong)(lock_stats[lbl].num_read_unlocked - lock_stats[lbl].num_read_locked),
+ (double)lock_stats[lbl].nsec_wait_for_read / 1000000.0,
+ lock_stats[lbl].num_read_locked ? ((double)lock_stats[lbl].nsec_wait_for_read / (double)lock_stats[lbl].num_read_locked) : 0);
+ }
+}
+
+void __ha_rwlock_init(struct ha_rwlock *l)
+{
+ memset(l, 0, sizeof(struct ha_rwlock));
+ __RWLOCK_INIT(&l->lock);
+}
+
+void __ha_rwlock_destroy(struct ha_rwlock *l)
+{
+ __RWLOCK_DESTROY(&l->lock);
+ memset(l, 0, sizeof(struct ha_rwlock));
+}
+
+
+void __ha_rwlock_wrlock(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_writers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_WRLOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_write, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_locked);
+
+ st->cur_writer = tbit;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_writers, ~tbit);
+}
+
+int __ha_rwlock_trywrlock(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+ int r;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ /* We set waiting writer because trywrlock could wait for readers to quit */
+ HA_ATOMIC_OR(&st->wait_writers, tbit);
+
+ start_time = now_mono_time();
+ r = __RWLOCK_TRYWRLOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_write, (now_mono_time() - start_time));
+ if (unlikely(r)) {
+ HA_ATOMIC_AND(&st->wait_writers, ~tbit);
+ return r;
+ }
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_locked);
+
+ st->cur_writer = tbit;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_writers, ~tbit);
+
+ return 0;
+}
+
+void __ha_rwlock_wrunlock(enum lock_label lbl,struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+
+ if (unlikely(!(st->cur_writer & tbit))) {
+ /* the thread is not owning the lock for write */
+ abort();
+ }
+
+ st->cur_writer = 0;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ __RWLOCK_WRUNLOCK(&l->lock);
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_unlocked);
+}
+
+void __ha_rwlock_rdlock(enum lock_label lbl,struct ha_rwlock *l)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_readers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_RDLOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_read, (now_mono_time() - start_time));
+ HA_ATOMIC_INC(&lock_stats[lbl].num_read_locked);
+
+ HA_ATOMIC_OR(&st->cur_readers, tbit);
+
+ HA_ATOMIC_AND(&st->wait_readers, ~tbit);
+}
+
+int __ha_rwlock_tryrdlock(enum lock_label lbl,struct ha_rwlock *l)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ int r;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ /* try read should never wait */
+ r = __RWLOCK_TRYRDLOCK(&l->lock);
+ if (unlikely(r))
+ return r;
+ HA_ATOMIC_INC(&lock_stats[lbl].num_read_locked);
+
+ HA_ATOMIC_OR(&st->cur_readers, tbit);
+
+ return 0;
+}
+
+void __ha_rwlock_rdunlock(enum lock_label lbl,struct ha_rwlock *l)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+
+ if (unlikely(!(st->cur_readers & tbit))) {
+ /* the thread is not owning the lock for read */
+ abort();
+ }
+
+ HA_ATOMIC_AND(&st->cur_readers, ~tbit);
+
+ __RWLOCK_RDUNLOCK(&l->lock);
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_read_unlocked);
+}
+
+void __ha_rwlock_wrtord(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_seeker) & tbit)
+ abort();
+
+ if (!(st->cur_writer & tbit))
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_readers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_WRTORD(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_read, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_read_locked);
+
+ HA_ATOMIC_OR(&st->cur_readers, tbit);
+ HA_ATOMIC_AND(&st->cur_writer, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_readers, ~tbit);
+}
+
+void __ha_rwlock_wrtosk(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_seeker) & tbit)
+ abort();
+
+ if (!(st->cur_writer & tbit))
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_seekers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_WRTOSK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_seek, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_seek_locked);
+
+ HA_ATOMIC_OR(&st->cur_seeker, tbit);
+ HA_ATOMIC_AND(&st->cur_writer, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_seekers, ~tbit);
+}
+
+void __ha_rwlock_sklock(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_seekers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_SKLOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_seek, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_seek_locked);
+
+ HA_ATOMIC_OR(&st->cur_seeker, tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_seekers, ~tbit);
+}
+
+void __ha_rwlock_sktowr(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_writer) & tbit)
+ abort();
+
+ if (!(st->cur_seeker & tbit))
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_writers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_SKTOWR(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_write, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_locked);
+
+ HA_ATOMIC_OR(&st->cur_writer, tbit);
+ HA_ATOMIC_AND(&st->cur_seeker, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_writers, ~tbit);
+}
+
+void __ha_rwlock_sktord(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if ((st->cur_readers | st->cur_writer) & tbit)
+ abort();
+
+ if (!(st->cur_seeker & tbit))
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_readers, tbit);
+
+ start_time = now_mono_time();
+ __RWLOCK_SKTORD(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_read, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_read_locked);
+
+ HA_ATOMIC_OR(&st->cur_readers, tbit);
+ HA_ATOMIC_AND(&st->cur_seeker, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->wait_readers, ~tbit);
+}
+
+void __ha_rwlock_skunlock(enum lock_label lbl,struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ if (!(st->cur_seeker & tbit))
+ abort();
+
+ HA_ATOMIC_AND(&st->cur_seeker, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ __RWLOCK_SKUNLOCK(&l->lock);
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_seek_unlocked);
+}
+
+int __ha_rwlock_trysklock(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+ int r;
+
+ if ((st->cur_readers | st->cur_seeker | st->cur_writer) & tbit)
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_seekers, tbit);
+
+ start_time = now_mono_time();
+ r = __RWLOCK_TRYSKLOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_seek, (now_mono_time() - start_time));
+
+ if (likely(!r)) {
+ /* got the lock ! */
+ HA_ATOMIC_INC(&lock_stats[lbl].num_seek_locked);
+ HA_ATOMIC_OR(&st->cur_seeker, tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+ }
+
+ HA_ATOMIC_AND(&st->wait_seekers, ~tbit);
+ return r;
+}
+
+int __ha_rwlock_tryrdtosk(enum lock_label lbl, struct ha_rwlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_rwlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+ int r;
+
+ if ((st->cur_writer | st->cur_seeker) & tbit)
+ abort();
+
+ if (!(st->cur_readers & tbit))
+ abort();
+
+ HA_ATOMIC_OR(&st->wait_seekers, tbit);
+
+ start_time = now_mono_time();
+ r = __RWLOCK_TRYRDTOSK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_seek, (now_mono_time() - start_time));
+
+ if (likely(!r)) {
+ /* got the lock ! */
+ HA_ATOMIC_INC(&lock_stats[lbl].num_seek_locked);
+ HA_ATOMIC_OR(&st->cur_seeker, tbit);
+ HA_ATOMIC_AND(&st->cur_readers, ~tbit);
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+ }
+
+ HA_ATOMIC_AND(&st->wait_seekers, ~tbit);
+ return r;
+}
+
+void __spin_init(struct ha_spinlock *l)
+{
+ memset(l, 0, sizeof(struct ha_spinlock));
+ __SPIN_INIT(&l->lock);
+}
+
+void __spin_destroy(struct ha_spinlock *l)
+{
+ __SPIN_DESTROY(&l->lock);
+ memset(l, 0, sizeof(struct ha_spinlock));
+}
+
+void __spin_lock(enum lock_label lbl, struct ha_spinlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_spinlock_state *st = &l->info.st[tgid-1];
+ uint64_t start_time;
+
+ if (unlikely(st->owner & tbit)) {
+ /* the thread is already owning the lock */
+ abort();
+ }
+
+ HA_ATOMIC_OR(&st->waiters, tbit);
+
+ start_time = now_mono_time();
+ __SPIN_LOCK(&l->lock);
+ HA_ATOMIC_ADD(&lock_stats[lbl].nsec_wait_for_write, (now_mono_time() - start_time));
+
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_locked);
+
+
+ st->owner = tbit;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ HA_ATOMIC_AND(&st->waiters, ~tbit);
+}
+
+int __spin_trylock(enum lock_label lbl, struct ha_spinlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_spinlock_state *st = &l->info.st[tgid-1];
+ int r;
+
+ if (unlikely(st->owner & tbit)) {
+ /* the thread is already owning the lock */
+ abort();
+ }
+
+ /* try read should never wait */
+ r = __SPIN_TRYLOCK(&l->lock);
+ if (unlikely(r))
+ return r;
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_locked);
+
+ st->owner = tbit;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ return 0;
+}
+
+void __spin_unlock(enum lock_label lbl, struct ha_spinlock *l,
+ const char *func, const char *file, int line)
+{
+ ulong tbit = (ti && ti->ltid_bit) ? ti->ltid_bit : 1;
+ struct ha_spinlock_state *st = &l->info.st[tgid-1];
+
+ if (unlikely(!(st->owner & tbit))) {
+ /* the thread is not owning the lock */
+ abort();
+ }
+
+ st->owner = 0;
+ l->info.last_location.function = func;
+ l->info.last_location.file = file;
+ l->info.last_location.line = line;
+
+ __SPIN_UNLOCK(&l->lock);
+ HA_ATOMIC_INC(&lock_stats[lbl].num_write_unlocked);
+}
+
+#endif // defined(DEBUG_THREAD) || defined(DEBUG_FULL)
+
+
+#if defined(USE_PTHREAD_EMULATION)
+
+/* pthread rwlock emulation using plocks (to avoid expensive futexes).
+ * these are a direct mapping on Progressive Locks, with the exception that
+ * since there's a common unlock operation in pthreads, we need to know if
+ * we need to unlock for reads or writes, so we set the topmost bit to 1 when
+ * a write lock is acquired to indicate that a write unlock needs to be
+ * performed. It's not a problem since this bit will never be used given that
+ * haproxy won't support as many threads as the plocks.
+ *
+ * The storage is the pthread_rwlock_t cast as an ulong
+ */
+
+int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr)
+{
+ ulong *lock = (ulong *)rwlock;
+
+ *lock = 0;
+ return 0;
+}
+
+int pthread_rwlock_destroy(pthread_rwlock_t *rwlock)
+{
+ ulong *lock = (ulong *)rwlock;
+
+ *lock = 0;
+ return 0;
+}
+
+int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock)
+{
+ pl_lorw_rdlock((unsigned long *)rwlock);
+ return 0;
+}
+
+int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock)
+{
+ return !!pl_cmpxchg((unsigned long *)rwlock, 0, PLOCK_LORW_SHR_BASE);
+}
+
+int pthread_rwlock_timedrdlock(pthread_rwlock_t *restrict rwlock, const struct timespec *restrict abstime)
+{
+ return pthread_rwlock_tryrdlock(rwlock);
+}
+
+int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock)
+{
+ pl_lorw_wrlock((unsigned long *)rwlock);
+ return 0;
+}
+
+int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock)
+{
+ return !!pl_cmpxchg((unsigned long *)rwlock, 0, PLOCK_LORW_EXC_BASE);
+}
+
+int pthread_rwlock_timedwrlock(pthread_rwlock_t *restrict rwlock, const struct timespec *restrict abstime)
+{
+ return pthread_rwlock_trywrlock(rwlock);
+}
+
+int pthread_rwlock_unlock(pthread_rwlock_t *rwlock)
+{
+ pl_lorw_unlock((unsigned long *)rwlock);
+ return 0;
+}
+#endif // defined(USE_PTHREAD_EMULATION)
+
+/* 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;
+ if (pthread_create(&dummy_thread, NULL, dummy_thread_function, NULL) == 0)
+ pthread_join(dummy_thread, NULL);
+}
+
+static void __thread_init(void)
+{
+ char *ptr = NULL;
+
+ preload_libgcc_s();
+
+ thread_cpus_enabled_at_boot = thread_cpus_enabled();
+ thread_cpus_enabled_at_boot = MIN(thread_cpus_enabled_at_boot, MAX_THREADS);
+
+ memprintf(&ptr, "Built with multi-threading support (MAX_TGROUPS=%d, MAX_THREADS=%d, default=%d).",
+ MAX_TGROUPS, 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
+}
+INITCALL0(STG_PREPARE, __thread_init);
+
+#else
+
+/* send signal <sig> to thread <thr> (send to process in fact) */
+void ha_tkill(unsigned int thr, int sig)
+{
+ raise(sig);
+}
+
+/* send signal <sig> to all threads (send to process in fact) */
+void ha_tkillall(int sig)
+{
+ raise(sig);
+}
+
+void ha_thread_relax(void)
+{
+#ifdef _POSIX_PRIORITY_SCHEDULING
+ sched_yield();
+#endif
+}
+
+REGISTER_BUILD_OPTS("Built without multi-threading support (USE_THREAD not set).");
+
+#endif // USE_THREAD
+
+
+/* Returns non-zero on anomaly (bound vs unbound), and emits a warning in this
+ * case.
+ */
+int thread_detect_binding_discrepancies(void)
+{
+#if defined(USE_CPU_AFFINITY)
+ uint th, tg, id;
+ uint tot_b = 0, tot_u = 0;
+ int first_b = -1;
+ int first_u = -1;
+
+ for (th = 0; th < global.nbthread; th++) {
+ tg = ha_thread_info[th].tgid;
+ id = ha_thread_info[th].ltid;
+
+ if (ha_cpuset_count(&cpu_map[tg - 1].thread[id]) == 0) {
+ tot_u++;
+ if (first_u < 0)
+ first_u = th;
+ } else {
+ tot_b++;
+ if (first_b < 0)
+ first_b = th;
+ }
+ }
+
+ if (tot_u > 0 && tot_b > 0) {
+ ha_warning("Found %u thread(s) mapped to a CPU and %u thread(s) not mapped to any CPU. "
+ "This will result in some threads being randomly assigned to the same CPU, "
+ "which will occasionally cause severe performance degradation. First thread "
+ "bound is %d and first thread not bound is %d. Please either bind all threads "
+ "or none (maybe some cpu-map directives are missing?).\n",
+ tot_b, tot_u, first_b, first_u);
+ return 1;
+ }
+#endif
+ return 0;
+}
+
+/* Returns non-zero on anomaly (more threads than CPUs), and emits a warning in
+ * this case. It checks against configured cpu-map if any, otherwise against
+ * the number of CPUs at boot if known. It's better to run it only after
+ * thread_detect_binding_discrepancies() so that mixed cases can be eliminated.
+ */
+int thread_detect_more_than_cpus(void)
+{
+#if defined(USE_CPU_AFFINITY)
+ struct hap_cpuset cpuset_map, cpuset_boot, cpuset_all;
+ uint th, tg, id;
+ int bound;
+ int tot_map, tot_all;
+
+ ha_cpuset_zero(&cpuset_boot);
+ ha_cpuset_zero(&cpuset_map);
+ ha_cpuset_zero(&cpuset_all);
+ bound = 0;
+ for (th = 0; th < global.nbthread; th++) {
+ tg = ha_thread_info[th].tgid;
+ id = ha_thread_info[th].ltid;
+ if (ha_cpuset_count(&cpu_map[tg - 1].thread[id])) {
+ ha_cpuset_or(&cpuset_map, &cpu_map[tg - 1].thread[id]);
+ bound++;
+ }
+ }
+
+ ha_cpuset_assign(&cpuset_all, &cpuset_map);
+ if (bound != global.nbthread) {
+ if (ha_cpuset_detect_bound(&cpuset_boot))
+ ha_cpuset_or(&cpuset_all, &cpuset_boot);
+ }
+
+ tot_map = ha_cpuset_count(&cpuset_map);
+ tot_all = ha_cpuset_count(&cpuset_all);
+
+ if (tot_map && bound > tot_map) {
+ ha_warning("This configuration binds %d threads to a total of %d CPUs via cpu-map "
+ "directives. This means that some threads will compete for the same CPU, "
+ "which will cause severe performance degradation. Please fix either the "
+ "'cpu-map' directives or set the global 'nbthread' value accordingly.\n",
+ bound, tot_map);
+ return 1;
+ }
+ else if (tot_all && global.nbthread > tot_all) {
+ ha_warning("This configuration enables %d threads running on a total of %d CPUs. "
+ "This means that some threads will compete for the same CPU, which will cause "
+ "severe performance degradation. Please either the 'cpu-map' directives to "
+ "adjust the CPUs to use, or fix the global 'nbthread' value.\n",
+ global.nbthread, tot_all);
+ return 1;
+ }
+#endif
+ return 0;
+}
+
+
+/* scans the configured thread mapping and establishes the final one. Returns <0
+ * on failure, >=0 on success.
+ */
+int thread_map_to_groups()
+{
+ int t, g, ut, ug;
+ int q, r;
+ ulong m __maybe_unused;
+
+ ut = ug = 0; // unassigned threads & groups
+
+ for (t = 0; t < global.nbthread; t++) {
+ if (!ha_thread_info[t].tg)
+ ut++;
+ }
+
+ for (g = 0; g < global.nbtgroups; g++) {
+ if (!ha_tgroup_info[g].count)
+ ug++;
+ ha_tgroup_info[g].tgid_bit = 1UL << g;
+ }
+
+ if (ug > ut) {
+ ha_alert("More unassigned thread-groups (%d) than threads (%d). Please reduce thread-groups\n", ug, ut);
+ return -1;
+ }
+
+ /* look for first unassigned thread */
+ for (t = 0; t < global.nbthread && ha_thread_info[t].tg; t++)
+ ;
+
+ /* assign threads to empty groups */
+ for (g = 0; ug && ut; ) {
+ /* due to sparse thread assignment we can end up with more threads
+ * per group on last assigned groups than former ones, so we must
+ * always try to pack the maximum remaining ones together first.
+ */
+ q = ut / ug;
+ r = ut % ug;
+ if ((q + !!r) > MAX_THREADS_PER_GROUP) {
+ ha_alert("Too many remaining unassigned threads (%d) for thread groups (%d). Please increase thread-groups or make sure to keep thread numbers contiguous\n", ut, ug);
+ return -1;
+ }
+
+ /* thread <t> is the next unassigned one. Let's look for next
+ * unassigned group, we know there are some left
+ */
+ while (ut >= ug && ha_tgroup_info[g].count)
+ g++;
+
+ /* group g is unassigned, try to fill it with consecutive threads */
+ while (ut && ut >= ug && ha_tgroup_info[g].count < q + !!r &&
+ (!ha_tgroup_info[g].count || t == ha_tgroup_info[g].base + ha_tgroup_info[g].count)) {
+
+ if (!ha_tgroup_info[g].count) {
+ /* assign new group */
+ ha_tgroup_info[g].base = t;
+ ug--;
+ }
+
+ ha_tgroup_info[g].count++;
+ ha_thread_info[t].tgid = g + 1;
+ ha_thread_info[t].tg = &ha_tgroup_info[g];
+ ha_thread_info[t].tg_ctx = &ha_tgroup_ctx[g];
+
+ ut--;
+ /* switch to next unassigned thread */
+ while (++t < global.nbthread && ha_thread_info[t].tg)
+ ;
+ }
+ }
+
+ if (ut) {
+ ha_alert("Remaining unassigned threads found (%d) because all groups are in use. Please increase 'thread-groups', reduce 'nbthreads' or remove or extend 'thread-group' enumerations.\n", ut);
+ return -1;
+ }
+
+ for (t = 0; t < global.nbthread; t++) {
+ ha_thread_info[t].tid = t;
+ ha_thread_info[t].ltid = t - ha_thread_info[t].tg->base;
+ ha_thread_info[t].ltid_bit = 1UL << ha_thread_info[t].ltid;
+ }
+
+ m = 0;
+ for (g = 0; g < global.nbtgroups; g++) {
+ ha_tgroup_info[g].threads_enabled = nbits(ha_tgroup_info[g].count);
+ /* for now, additional threads are not started, so we should
+ * consider them as harmless and idle.
+ * This will get automatically updated when such threads are
+ * started in run_thread_poll_loop()
+ * Without this, thread_isolate() and thread_isolate_full()
+ * will fail to work as long as secondary threads did not enter
+ * the polling loop at least once.
+ */
+ ha_tgroup_ctx[g].threads_harmless = ha_tgroup_info[g].threads_enabled;
+ ha_tgroup_ctx[g].threads_idle = ha_tgroup_info[g].threads_enabled;
+ if (!ha_tgroup_info[g].count)
+ continue;
+ m |= 1UL << g;
+
+ }
+
+#ifdef USE_THREAD
+ all_tgroups_mask = m;
+#endif
+ return 0;
+}
+
+/* Converts a configuration thread set based on either absolute or relative
+ * thread numbers into a global group+mask. This is essentially for use with
+ * the "thread" directive on "bind" lines, where "thread 4-6,10-12" might be
+ * turned to "2/1-3,4/1-3". It cannot be used before the thread mapping above
+ * was completed and the thread group numbers configured. The thread_set is
+ * replaced by the resolved group-based one. It is possible to force a single
+ * default group for unspecified sets instead of enabling all groups by passing
+ * this group's non-zero value to defgrp.
+ *
+ * Returns <0 on failure, >=0 on success.
+ */
+int thread_resolve_group_mask(struct thread_set *ts, int defgrp, char **err)
+{
+ struct thread_set new_ts = { };
+ ulong mask, imask;
+ uint g;
+
+ if (!ts->grps) {
+ /* unspecified group, IDs are global */
+ if (thread_set_is_empty(ts)) {
+ /* all threads of all groups, unless defgrp is set and
+ * we then set it as the only group.
+ */
+ for (g = defgrp ? defgrp-1 : 0; g < (defgrp ? defgrp : global.nbtgroups); g++) {
+ new_ts.rel[g] = ha_tgroup_info[g].threads_enabled;
+ if (new_ts.rel[g])
+ new_ts.grps |= 1UL << g;
+ }
+ } else {
+ /* some absolute threads are set, we must remap them to
+ * relative ones. Each group cannot have more than
+ * LONGBITS threads, thus it spans at most two absolute
+ * blocks.
+ */
+ for (g = 0; g < global.nbtgroups; g++) {
+ uint block = ha_tgroup_info[g].base / LONGBITS;
+ uint base = ha_tgroup_info[g].base % LONGBITS;
+
+ mask = ts->abs[block] >> base;
+ if (base &&
+ (block + 1) < sizeof(ts->abs) / sizeof(ts->abs[0]) &&
+ ha_tgroup_info[g].count > (LONGBITS - base))
+ mask |= ts->abs[block + 1] << (LONGBITS - base);
+ mask &= nbits(ha_tgroup_info[g].count);
+ mask &= ha_tgroup_info[g].threads_enabled;
+
+ /* now the mask exactly matches the threads to be enabled
+ * in this group.
+ */
+ new_ts.rel[g] |= mask;
+ if (new_ts.rel[g])
+ new_ts.grps |= 1UL << g;
+ }
+ }
+ } else {
+ /* groups were specified */
+ for (g = 0; g < MAX_TGROUPS; g++) {
+ imask = ts->rel[g];
+ if (!imask)
+ continue;
+
+ if (g >= global.nbtgroups) {
+ memprintf(err, "'thread' directive references non-existing thread group %u", g+1);
+ return -1;
+ }
+
+ /* some relative threads are set. Keep only existing ones for this group */
+ mask = nbits(ha_tgroup_info[g].count);
+
+ if (!(mask & imask)) {
+ /* no intersection between the thread group's
+ * threads and the bind line's.
+ */
+#ifdef THREAD_AUTO_ADJUST_GROUPS
+ unsigned long new_mask = 0;
+
+ while (imask) {
+ new_mask |= imask & mask;
+ imask >>= ha_tgroup_info[g].count;
+ }
+ imask = new_mask;
+#else
+ memprintf(err, "'thread' directive only references threads not belonging to group %u", g+1);
+ return -1;
+#endif
+ }
+
+ new_ts.rel[g] = imask & mask;
+ if (new_ts.rel[g])
+ new_ts.grps |= 1UL << g;
+ }
+ }
+
+ /* update the thread_set */
+ if (!thread_set_nth_group(&new_ts, 0)) {
+ memprintf(err, "'thread' directive only references non-existing threads");
+ return -1;
+ }
+
+ *ts = new_ts;
+ return 0;
+}
+
+/* Parse a string representing a thread set in one of the following forms:
+ *
+ * - { "all" | "odd" | "even" | <abs_num> [ "-" <abs_num> ] }[,...]
+ * => these are (lists of) absolute thread numbers
+ *
+ * - <tgnum> "/" { "all" | "odd" | "even" | <rel_num> [ "-" <rel_num> ][,...]
+ * => these are (lists of) per-group relative thread numbers. All numbers
+ * must be lower than or equal to LONGBITS. When multiple list elements
+ * are provided, each of them must contain the thread group number.
+ *
+ * Minimum value for a thread or group number is always 1. Maximum value for an
+ * absolute thread number is MAX_THREADS, maximum value for a relative thread
+ * number is MAX_THREADS_PER_GROUP, an maximum value for a thread group is
+ * MAX_TGROUPS. "all", "even" and "odd" will be bound by MAX_THREADS and/or
+ * MAX_THREADS_PER_GROUP in any case. In ranges, a missing digit before "-"
+ * is implicitly 1, and a missing digit after "-" is implicitly the highest of
+ * its class. As such "-" is equivalent to "all", allowing to build strings
+ * such as "${MIN}-${MAX}" where both MIN and MAX are optional.
+ *
+ * It is not valid to mix absolute and relative numbers. As such:
+ * - all valid (all absolute threads)
+ * - 12-19,24-31 valid (abs threads 12 to 19 and 24 to 31)
+ * - 1/all valid (all 32 or 64 threads of group 1)
+ * - 1/1-4,1/8-10,2/1 valid
+ * - 1/1-4,8-10 invalid (mixes relatve "1/1-4" with absolute "8-10")
+ * - 1-4,8-10,2/1 invalid (mixes absolute "1-4,8-10" with relative "2/1")
+ * - 1/odd-4 invalid (mixes range with boundary)
+ *
+ * The target thread set is *completed* with supported threads, which means
+ * that it's the caller's responsibility for pre-initializing it. If the target
+ * thread set is NULL, it's not updated and the function only verifies that the
+ * input parses.
+ *
+ * On success, it returns 0. otherwise it returns non-zero with an error
+ * message in <err>.
+ */
+int parse_thread_set(const char *arg, struct thread_set *ts, char **err)
+{
+ const char *set;
+ const char *sep;
+ int v, min, max, tg;
+ int is_rel;
+
+ /* search for the first delimiter (',', '-' or '/') to decide whether
+ * we're facing an absolute or relative form. The relative form always
+ * starts with a number followed by a slash.
+ */
+ for (sep = arg; isdigit((uchar)*sep); sep++)
+ ;
+
+ is_rel = (/*sep > arg &&*/ *sep == '/'); /* relative form */
+
+ /* from there we have to cut the thread spec around commas */
+
+ set = arg;
+ tg = 0;
+ while (*set) {
+ /* note: we can't use strtol() here because "-3" would parse as
+ * (-3) while we want to stop before the "-", so we find the
+ * separator ourselves and rely on atoi() whose value we may
+ * ignore depending where the separator is.
+ */
+ for (sep = set; isdigit((uchar)*sep); sep++)
+ ;
+
+ if (sep != set && *sep && *sep != '/' && *sep != '-' && *sep != ',') {
+ memprintf(err, "invalid character '%c' in thread set specification: '%s'.", *sep, set);
+ return -1;
+ }
+
+ v = (sep != set) ? atoi(set) : 0;
+
+ /* Now we know that the string is made of an optional series of digits
+ * optionally followed by one of the delimiters above, or that it
+ * starts with a different character.
+ */
+
+ /* first, let's search for the thread group (digits before '/') */
+
+ if (tg || !is_rel) {
+ /* thread group already specified or not expected if absolute spec */
+ if (*sep == '/') {
+ if (tg)
+ memprintf(err, "redundant thread group specification '%s' for group %d", set, tg);
+ else
+ memprintf(err, "group-relative thread specification '%s' is not permitted after a absolute thread range.", set);
+ return -1;
+ }
+ } else {
+ /* this is a group-relative spec, first field is the group number */
+ if (sep == set && *sep == '/') {
+ memprintf(err, "thread group number expected before '%s'.", set);
+ return -1;
+ }
+
+ if (*sep != '/') {
+ memprintf(err, "absolute thread specification '%s' is not permitted after a group-relative thread range.", set);
+ return -1;
+ }
+
+ if (v < 1 || v > MAX_TGROUPS) {
+ memprintf(err, "invalid thread group number '%d', permitted range is 1..%d in '%s'.", v, MAX_TGROUPS, set);
+ return -1;
+ }
+
+ tg = v;
+
+ /* skip group number and go on with set,sep,v as if
+ * there was no group number.
+ */
+ set = sep + 1;
+ continue;
+ }
+
+ /* Now 'set' starts at the min thread number, whose value is in v if any,
+ * and preset the max to it, unless the range is filled at once via "all"
+ * (stored as 1:0), "odd" (stored as) 1:-1, or "even" (stored as 1:-2).
+ * 'sep' points to the next non-digit which may be set itself e.g. for
+ * "all" etc or "-xx".
+ */
+
+ if (!*set) {
+ /* empty set sets no restriction */
+ min = 1;
+ max = is_rel ? MAX_THREADS_PER_GROUP : MAX_THREADS;
+ }
+ else {
+ if (sep != set && *sep && *sep != '-' && *sep != ',') {
+ // Only delimiters are permitted around digits.
+ memprintf(err, "invalid character '%c' in thread set specification: '%s'.", *sep, set);
+ return -1;
+ }
+
+ /* for non-digits, find next delim */
+ for (; *sep && *sep != '-' && *sep != ','; sep++)
+ ;
+
+ min = max = 1;
+ if (sep != set) {
+ /* non-empty first thread */
+ if (isteq(ist2(set, sep-set), ist("all")))
+ max = 0;
+ else if (isteq(ist2(set, sep-set), ist("odd")))
+ max = -1;
+ else if (isteq(ist2(set, sep-set), ist("even")))
+ max = -2;
+ else if (v)
+ min = max = v;
+ else
+ max = min = 0; // throw an error below
+ }
+
+ if (min < 1 || min > MAX_THREADS || (is_rel && min > MAX_THREADS_PER_GROUP)) {
+ memprintf(err, "invalid first thread number '%s', permitted range is 1..%d, or 'all', 'odd', 'even'.",
+ set, is_rel ? MAX_THREADS_PER_GROUP : MAX_THREADS);
+ return -1;
+ }
+
+ /* is this a range ? */
+ if (*sep == '-') {
+ if (min != max) {
+ memprintf(err, "extraneous range after 'all', 'odd' or 'even': '%s'.", set);
+ return -1;
+ }
+
+ /* this is a seemingly valid range, there may be another number */
+ for (set = ++sep; isdigit((uchar)*sep); sep++)
+ ;
+ v = atoi(set);
+
+ if (sep == set) { // no digit: to the max
+ max = is_rel ? MAX_THREADS_PER_GROUP : MAX_THREADS;
+ if (*sep && *sep != ',')
+ max = 0; // throw an error below
+ } else
+ max = v;
+
+ if (max < 1 || max > MAX_THREADS || (is_rel && max > MAX_THREADS_PER_GROUP)) {
+ memprintf(err, "invalid last thread number '%s', permitted range is 1..%d.",
+ set, is_rel ? MAX_THREADS_PER_GROUP : MAX_THREADS);
+ return -1;
+ }
+ }
+
+ /* here sep points to the first non-digit after the thread spec,
+ * must be a valid delimiter.
+ */
+ if (*sep && *sep != ',') {
+ memprintf(err, "invalid character '%c' after thread set specification: '%s'.", *sep, set);
+ return -1;
+ }
+ }
+
+ /* store values */
+ if (ts) {
+ if (is_rel) {
+ /* group-relative thread numbers */
+ ts->grps |= 1UL << (tg - 1);
+
+ if (max >= min) {
+ for (v = min; v <= max; v++)
+ ts->rel[tg - 1] |= 1UL << (v - 1);
+ } else {
+ memset(&ts->rel[tg - 1],
+ (max == 0) ? 0xff /* all */ : (max == -1) ? 0x55 /* odd */: 0xaa /* even */,
+ sizeof(ts->rel[tg - 1]));
+ }
+ } else {
+ /* absolute thread numbers */
+ if (max >= min) {
+ for (v = min; v <= max; v++)
+ ts->abs[(v - 1) / LONGBITS] |= 1UL << ((v - 1) % LONGBITS);
+ } else {
+ memset(&ts->abs,
+ (max == 0) ? 0xff /* all */ : (max == -1) ? 0x55 /* odd */: 0xaa /* even */,
+ sizeof(ts->abs));
+ }
+ }
+ }
+
+ set = *sep ? sep + 1 : sep;
+ tg = 0;
+ }
+ return 0;
+}
+
+/* Parse the "nbthread" global directive, which takes an integer argument that
+ * contains the desired number of threads.
+ */
+static int cfg_parse_nbthread(char **args, int section_type, struct proxy *curpx,
+ const struct proxy *defpx, const char *file, int line,
+ char **err)
+{
+ long nbthread;
+ char *errptr;
+
+ if (too_many_args(1, args, err, NULL))
+ return -1;
+
+ if (non_global_section_parsed == 1) {
+ memprintf(err, "'%s' not allowed if a non-global section was previously defined. This parameter must be declared in the first global section", args[0]);
+ return -1;
+ }
+
+ nbthread = strtol(args[1], &errptr, 10);
+ if (!*args[1] || *errptr) {
+ memprintf(err, "'%s' passed a missing or unparsable integer value in '%s'", args[0], args[1]);
+ return -1;
+ }
+
+#ifndef USE_THREAD
+ if (nbthread != 1) {
+ memprintf(err, "'%s' specified with a value other than 1 while HAProxy is not compiled with threads support. Please check build options for USE_THREAD", args[0]);
+ return -1;
+ }
+#else
+ if (nbthread < 1 || nbthread > MAX_THREADS) {
+ memprintf(err, "'%s' value must be between 1 and %d (was %ld)", args[0], MAX_THREADS, nbthread);
+ return -1;
+ }
+#endif
+
+ HA_DIAG_WARNING_COND(global.nbthread,
+ "parsing [%s:%d] : '%s' is already defined and will be overridden.\n",
+ file, line, args[0]);
+
+ global.nbthread = nbthread;
+ return 0;
+}
+
+/* Parse the "thread-group" global directive, which takes an integer argument
+ * that designates a thread group, and a list of threads to put into that group.
+ */
+static int cfg_parse_thread_group(char **args, int section_type, struct proxy *curpx,
+ const struct proxy *defpx, const char *file, int line,
+ char **err)
+{
+ char *errptr;
+ long tnum, tend, tgroup;
+ int arg, tot;
+
+ if (non_global_section_parsed == 1) {
+ memprintf(err, "'%s' not allowed if a non-global section was previously defined. This parameter must be declared in the first global section", args[0]);
+ return -1;
+ }
+
+ tgroup = strtol(args[1], &errptr, 10);
+ if (!*args[1] || *errptr) {
+ memprintf(err, "'%s' passed a missing or unparsable integer value in '%s'", args[0], args[1]);
+ return -1;
+ }
+
+ if (tgroup < 1 || tgroup > MAX_TGROUPS) {
+ memprintf(err, "'%s' thread-group number must be between 1 and %d (was %ld)", args[0], MAX_TGROUPS, tgroup);
+ return -1;
+ }
+
+ /* look for a preliminary definition of any thread pointing to this
+ * group, and remove them.
+ */
+ if (ha_tgroup_info[tgroup-1].count) {
+ ha_warning("parsing [%s:%d] : '%s %ld' was already defined and will be overridden.\n",
+ file, line, args[0], tgroup);
+
+ for (tnum = ha_tgroup_info[tgroup-1].base;
+ tnum < ha_tgroup_info[tgroup-1].base + ha_tgroup_info[tgroup-1].count;
+ tnum++) {
+ if (ha_thread_info[tnum-1].tg == &ha_tgroup_info[tgroup-1]) {
+ ha_thread_info[tnum-1].tg = NULL;
+ ha_thread_info[tnum-1].tgid = 0;
+ ha_thread_info[tnum-1].tg_ctx = NULL;
+ }
+ }
+ ha_tgroup_info[tgroup-1].count = ha_tgroup_info[tgroup-1].base = 0;
+ }
+
+ tot = 0;
+ for (arg = 2; args[arg] && *args[arg]; arg++) {
+ tend = tnum = strtol(args[arg], &errptr, 10);
+
+ if (*errptr == '-')
+ tend = strtol(errptr + 1, &errptr, 10);
+
+ if (*errptr || tnum < 1 || tend < 1 || tnum > MAX_THREADS || tend > MAX_THREADS) {
+ memprintf(err, "'%s %ld' passed an unparsable or invalid thread number '%s' (valid range is 1 to %d)", args[0], tgroup, args[arg], MAX_THREADS);
+ return -1;
+ }
+
+ for(; tnum <= tend; tnum++) {
+ if (ha_thread_info[tnum-1].tg == &ha_tgroup_info[tgroup-1]) {
+ ha_warning("parsing [%s:%d] : '%s %ld': thread %ld assigned more than once on the same line.\n",
+ file, line, args[0], tgroup, tnum);
+ } else if (ha_thread_info[tnum-1].tg) {
+ ha_warning("parsing [%s:%d] : '%s %ld': thread %ld was previously assigned to thread group %ld and will be overridden.\n",
+ file, line, args[0], tgroup, tnum,
+ (long)(ha_thread_info[tnum-1].tg - &ha_tgroup_info[0] + 1));
+ }
+
+ if (!ha_tgroup_info[tgroup-1].count) {
+ ha_tgroup_info[tgroup-1].base = tnum-1;
+ ha_tgroup_info[tgroup-1].count = 1;
+ }
+ else if (tnum >= ha_tgroup_info[tgroup-1].base + ha_tgroup_info[tgroup-1].count) {
+ ha_tgroup_info[tgroup-1].count = tnum - ha_tgroup_info[tgroup-1].base;
+ }
+ else if (tnum < ha_tgroup_info[tgroup-1].base) {
+ ha_tgroup_info[tgroup-1].count += ha_tgroup_info[tgroup-1].base - tnum-1;
+ ha_tgroup_info[tgroup-1].base = tnum - 1;
+ }
+
+ ha_thread_info[tnum-1].tgid = tgroup;
+ ha_thread_info[tnum-1].tg = &ha_tgroup_info[tgroup-1];
+ ha_thread_info[tnum-1].tg_ctx = &ha_tgroup_ctx[tgroup-1];
+ tot++;
+ }
+ }
+
+ if (ha_tgroup_info[tgroup-1].count > tot) {
+ memprintf(err, "'%s %ld' assigned sparse threads, only contiguous supported", args[0], tgroup);
+ return -1;
+ }
+
+ if (ha_tgroup_info[tgroup-1].count > MAX_THREADS_PER_GROUP) {
+ memprintf(err, "'%s %ld' assigned too many threads (%d, max=%d)", args[0], tgroup, tot, MAX_THREADS_PER_GROUP);
+ return -1;
+ }
+
+ return 0;
+}
+
+/* Parse the "thread-groups" global directive, which takes an integer argument
+ * that contains the desired number of thread groups.
+ */
+static int cfg_parse_thread_groups(char **args, int section_type, struct proxy *curpx,
+ const struct proxy *defpx, const char *file, int line,
+ char **err)
+{
+ long nbtgroups;
+ char *errptr;
+
+ if (too_many_args(1, args, err, NULL))
+ return -1;
+
+ if (non_global_section_parsed == 1) {
+ memprintf(err, "'%s' not allowed if a non-global section was previously defined. This parameter must be declared in the first global section", args[0]);
+ return -1;
+ }
+
+ nbtgroups = strtol(args[1], &errptr, 10);
+ if (!*args[1] || *errptr) {
+ memprintf(err, "'%s' passed a missing or unparsable integer value in '%s'", args[0], args[1]);
+ return -1;
+ }
+
+#ifndef USE_THREAD
+ if (nbtgroups != 1) {
+ memprintf(err, "'%s' specified with a value other than 1 while HAProxy is not compiled with threads support. Please check build options for USE_THREAD", args[0]);
+ return -1;
+ }
+#else
+ if (nbtgroups < 1 || nbtgroups > MAX_TGROUPS) {
+ memprintf(err, "'%s' value must be between 1 and %d (was %ld)", args[0], MAX_TGROUPS, nbtgroups);
+ return -1;
+ }
+#endif
+
+ HA_DIAG_WARNING_COND(global.nbtgroups,
+ "parsing [%s:%d] : '%s' is already defined and will be overridden.\n",
+ file, line, args[0]);
+
+ global.nbtgroups = nbtgroups;
+ return 0;
+}
+
+/* config keyword parsers */
+static struct cfg_kw_list cfg_kws = {ILH, {
+ { CFG_GLOBAL, "nbthread", cfg_parse_nbthread, 0 },
+ { CFG_GLOBAL, "thread-group", cfg_parse_thread_group, 0 },
+ { CFG_GLOBAL, "thread-groups", cfg_parse_thread_groups, 0 },
+ { 0, NULL, NULL }
+}};
+
+INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws);