blob: 59bfbf89cefd3dc7408468be4f21efe1daf4496b [file] [log] [blame]
/*
* 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.
*/
while (1) {
for (tgrp = 0; tgrp < global.nbtgroups; tgrp++) {
while ((_HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_harmless) &
ha_tgroup_info[tgrp].threads_enabled) != ha_tgroup_info[tgrp].threads_enabled)
ha_thread_relax();
}
/* Now we've seen all threads marked harmless, we can try to run
* by competing with other threads to win the race of the isolated
* thread. It eventually converges since winners will enventually
* relax their request and go back to wait for this to be over.
* Competing on this only after seeing all threads harmless limits
* the write contention.
*/
thr = _HA_ATOMIC_LOAD(&isolated_thread);
if (thr == ~0U && _HA_ATOMIC_CAS(&isolated_thread, &thr, tid))
break; // we won!
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.
*/
while (1) {
for (tgrp = 0; tgrp < global.nbtgroups; tgrp++) {
while ((_HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_harmless) &
_HA_ATOMIC_LOAD(&ha_tgroup_ctx[tgrp].threads_idle) &
ha_tgroup_info[tgrp].threads_enabled) != ha_tgroup_info[tgrp].threads_enabled)
ha_thread_relax();
}
/* Now we've seen all threads marked harmless and idle, we can
* try to run by competing with other threads to win the race
* of the isolated thread. It eventually converges since winners
* will enventually relax their request and go back to wait for
* this to be over. Competing on this only after seeing all
* threads harmless+idle limits the write contention.
*/
thr = _HA_ATOMIC_LOAD(&isolated_thread);
if (thr == ~0U && _HA_ATOMIC_CAS(&isolated_thread, &thr, tid))
break; // we won!
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);
}
/* 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].proc))
ha_cpuset_and(&cpu_map[tgid - 1].thread[ti->ltid], &cpu_map[tgid - 1].proc);
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 */
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);
ret = MIN(ret, MAX_THREADS);
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 LOGSRV_LOCK: return "LOGSRV";
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 QUIC_LOCK: return "QUIC";
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 : %lu\n"
"\t # write unlock: %lu (%ld)\n"
"\t # wait time for write : %.3f msec\n"
"\t # wait time for write/lock: %.3f nsec\n",
lock_stats[lbl].num_write_locked,
lock_stats[lbl].num_write_unlocked,
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 : %lu\n"
"\t # seek unlock : %lu (%ld)\n"
"\t # wait time for seek : %.3f msec\n"
"\t # wait time for seek/lock : %.3f nsec\n",
lock_stats[lbl].num_seek_locked,
lock_stats[lbl].num_seek_unlocked,
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 : %lu\n"
"\t # read unlock : %lu (%ld)\n"
"\t # wait time for read : %.3f msec\n"
"\t # wait time for read/lock : %.3f nsec\n",
lock_stats[lbl].num_read_locked,
lock_stats[lbl].num_read_unlocked,
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;
pthread_create(&dummy_thread, NULL, dummy_thread_function, NULL);
pthread_join(dummy_thread, NULL);
}
static void __thread_init(void)
{
char *ptr = NULL;
preload_libgcc_s();
thread_cpus_enabled_at_boot = thread_cpus_enabled();
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
/* 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", ug, ut);
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);
if (!ha_tgroup_info[g].count)
continue;
m |= 1UL << g;
}
#ifdef USE_THREAD
all_tgroups_mask = m;
#endif
return 0;
}
/* converts a configuration thread num or group+mask to a global group+mask
* depending on the configured thread group id. This is essentially for use
* with the "thread" directive on "bind" lines, where "thread 4-6" might be
* turned to "2/1-3". It cannot be used before the thread mapping above was
* completed and the thread group number configured. Possible options:
* - igid == 0: imask represents global IDs. We have to check that all
* configured threads in the mask belong to the same group. If imask is zero
* it means everything, so for now we only support this with a single group.
* - igid > 0, imask = 0: convert global values to local values for this thread
* - igid > 0, imask > 0: convert global values to local values
* Note that the output mask is always local to the group.
*
* Returns <0 on failure, >=0 on success.
*/
int thread_resolve_group_mask(uint igid, ulong imask, uint *ogid, ulong *omask, char **err)
{
ulong mask;
uint t;
if (igid == 0) {
/* unspecified group, IDs are global */
if (!imask) {
/* all threads of all groups */
if (global.nbtgroups > 1) {
memprintf(err, "'thread' directive spans multiple groups");
return -1;
}
*ogid = 1; // first and only group
*omask = ha_tgroup_info[0].threads_enabled;
return 0;
} else {
/* some global threads */
for (t = 0; t < global.nbthread; t++) {
if (imask & (1UL << t)) {
if (ha_thread_info[t].tgid != igid) {
if (!igid)
igid = ha_thread_info[t].tgid;
else {
memprintf(err, "'thread' directive spans multiple groups (at least %u and %u)", igid, ha_thread_info[t].tgid);
return -1;
}
}
}
}
if (!igid) {
memprintf(err, "'thread' directive contains threads that belong to no group");
return -1;
}
/* we have a valid group, convert this to global thread IDs */
*ogid = igid;
imask = imask >> ha_tgroup_info[igid - 1].base;
imask &= ha_tgroup_info[igid - 1].threads_enabled;
*omask = imask;
return 0;
}
} else {
/* group was specified */
if (igid > global.nbtgroups) {
memprintf(err, "'thread' directive references non-existing thread group %u", igid);
return -1;
}
if (!imask) {
/* all threads of this groups. Let's make a mask from their count and base. */
*ogid = igid;
*omask = nbits(ha_tgroup_info[igid - 1].count);
return 0;
} else {
/* some local threads. Keep only existing ones for this group */
mask = nbits(ha_tgroup_info[igid - 1].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[igid - 1].count;
}
imask = new_mask;
#else
memprintf(err, "'thread' directive only references threads not belonging to the group");
return -1;
#endif
}
*omask = mask & imask;
*ogid = igid;
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);