Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 1 | /* |
| 2 | * functions about threads. |
| 3 | * |
| 4 | * Copyright (C) 2017 Christopher Fauet - cfaulet@haproxy.com |
| 5 | * |
| 6 | * This program is free software; you can redistribute it and/or |
| 7 | * modify it under the terms of the GNU General Public License |
| 8 | * as published by the Free Software Foundation; either version |
| 9 | * 2 of the License, or (at your option) any later version. |
| 10 | * |
| 11 | */ |
| 12 | |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 13 | #include <unistd.h> |
Willy Tarreau | 0ccd322 | 2018-07-30 10:34:35 +0200 | [diff] [blame] | 14 | #include <stdlib.h> |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 15 | #include <fcntl.h> |
| 16 | |
Willy Tarreau | 0493149 | 2017-11-03 23:39:25 +0100 | [diff] [blame] | 17 | #include <common/cfgparse.h> |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 18 | #include <common/hathreads.h> |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 19 | #include <common/standard.h> |
| 20 | #include <proto/fd.h> |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 21 | |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 22 | |
Willy Tarreau | e96e61c | 2018-03-29 18:54:33 +0200 | [diff] [blame] | 23 | /* Dummy I/O handler used by the sync pipe.*/ |
| 24 | void thread_sync_io_handler(int fd) |
| 25 | { |
| 26 | } |
| 27 | |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 28 | #ifdef USE_THREAD |
| 29 | |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 30 | static HA_SPINLOCK_T sync_lock; |
| 31 | static int threads_sync_pipe[2]; |
| 32 | static unsigned long threads_want_sync = 0; |
Willy Tarreau | 60b639c | 2018-08-02 10:16:17 +0200 | [diff] [blame] | 33 | volatile unsigned long threads_want_rdv_mask = 0; |
| 34 | volatile unsigned long threads_harmless_mask = 0; |
Willy Tarreau | 0ccd322 | 2018-07-30 10:34:35 +0200 | [diff] [blame] | 35 | volatile unsigned long all_threads_mask = 1; // nbthread 1 assumed by default |
Willy Tarreau | 0c026f4 | 2018-08-01 19:12:20 +0200 | [diff] [blame] | 36 | THREAD_LOCAL unsigned int tid = 0; |
| 37 | THREAD_LOCAL unsigned long tid_bit = (1UL << 0); |
| 38 | |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 39 | |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 40 | #if defined(DEBUG_THREAD) || defined(DEBUG_FULL) |
| 41 | struct lock_stat lock_stats[LOCK_LABELS]; |
| 42 | #endif |
| 43 | |
Olivier Houchard | 3e12304 | 2018-07-27 17:06:59 +0200 | [diff] [blame] | 44 | /* Initializes the sync point. It creates a pipe used by threads to wake up all |
| 45 | * others when a sync is requested. It also initializes the mask of all created |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 46 | * threads. It returns 0 on success and -1 if an error occurred. |
| 47 | */ |
Willy Tarreau | 0ccd322 | 2018-07-30 10:34:35 +0200 | [diff] [blame] | 48 | int thread_sync_init() |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 49 | { |
| 50 | int rfd; |
| 51 | |
| 52 | if (pipe(threads_sync_pipe) < 0) |
| 53 | return -1; |
| 54 | |
| 55 | rfd = threads_sync_pipe[0]; |
| 56 | fcntl(rfd, F_SETFL, O_NONBLOCK); |
Willy Tarreau | a9786b6 | 2018-01-25 07:22:13 +0100 | [diff] [blame] | 57 | fd_insert(rfd, thread_sync_io_handler, thread_sync_io_handler, MAX_THREADS_MASK); |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 58 | return 0; |
| 59 | } |
| 60 | |
| 61 | /* Enables the sync point. */ |
| 62 | void thread_sync_enable(void) |
| 63 | { |
| 64 | fd_want_recv(threads_sync_pipe[0]); |
| 65 | } |
| 66 | |
| 67 | /* Called when a thread want to pass into the sync point. It subscribes the |
| 68 | * current thread in threads waiting for sync by update a bit-field. It this is |
| 69 | * the first one, it wakeup all other threads by writing on the sync pipe. |
| 70 | */ |
| 71 | void thread_want_sync() |
| 72 | { |
Willy Tarreau | ab657ce | 2018-08-07 10:07:15 +0200 | [diff] [blame^] | 73 | if (all_threads_mask & (all_threads_mask - 1)) { |
Christopher Faulet | 81991d3 | 2017-12-02 09:53:24 +0100 | [diff] [blame] | 74 | if (threads_want_sync & tid_bit) |
| 75 | return; |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 76 | if (HA_ATOMIC_OR(&threads_want_sync, tid_bit) == tid_bit) |
| 77 | shut_your_big_mouth_gcc(write(threads_sync_pipe[1], "S", 1)); |
| 78 | } |
| 79 | else { |
| 80 | threads_want_sync = 1; |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | /* Returns 1 if no thread has requested a sync. Otherwise, it returns 0. */ |
| 85 | int thread_no_sync() |
| 86 | { |
Christopher Faulet | 148b16e | 2018-05-02 16:58:40 +0200 | [diff] [blame] | 87 | return (threads_want_sync == 0UL); |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 88 | } |
| 89 | |
| 90 | /* Returns 1 if the current thread has requested a sync. Otherwise, it returns |
| 91 | * 0. |
| 92 | */ |
| 93 | int thread_need_sync() |
| 94 | { |
Christopher Faulet | 148b16e | 2018-05-02 16:58:40 +0200 | [diff] [blame] | 95 | return ((threads_want_sync & tid_bit) != 0UL); |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 96 | } |
| 97 | |
| 98 | /* Thread barrier. Synchronizes all threads at the barrier referenced by |
| 99 | * <barrier>. The calling thread shall block until all other threads have called |
| 100 | * thread_sync_barrier specifying the same barrier. |
| 101 | * |
| 102 | * If you need to use several barriers at differnt points, you need to use a |
| 103 | * different <barrier> for each point. |
| 104 | */ |
| 105 | static inline void thread_sync_barrier(volatile unsigned long *barrier) |
| 106 | { |
| 107 | unsigned long old = all_threads_mask; |
| 108 | |
| 109 | HA_ATOMIC_CAS(barrier, &old, 0); |
Christopher Faulet | 209d02a | 2017-10-27 23:01:38 +0200 | [diff] [blame] | 110 | HA_ATOMIC_OR(barrier, tid_bit); |
Willy Tarreau | 3ea2490 | 2018-07-27 07:47:24 +0200 | [diff] [blame] | 111 | |
| 112 | /* Note below: we need to wait for all threads to join here, but in |
| 113 | * case several threads are scheduled on the same CPU, busy polling |
| 114 | * will instead degrade the performance, forcing other threads to |
| 115 | * wait longer (typically in epoll_wait()). Let's use sched_yield() |
| 116 | * when available instead. |
| 117 | */ |
| 118 | while ((*barrier & all_threads_mask) != all_threads_mask) { |
| 119 | #if _POSIX_PRIORITY_SCHEDULING |
| 120 | sched_yield(); |
| 121 | #else |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 122 | pl_cpu_relax(); |
Willy Tarreau | 3ea2490 | 2018-07-27 07:47:24 +0200 | [diff] [blame] | 123 | #endif |
| 124 | } |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 125 | } |
| 126 | |
| 127 | /* Enter into the sync point and lock it if the current thread has requested a |
| 128 | * sync. */ |
| 129 | void thread_enter_sync() |
| 130 | { |
| 131 | static volatile unsigned long barrier = 0; |
| 132 | |
Willy Tarreau | 0c026f4 | 2018-08-01 19:12:20 +0200 | [diff] [blame] | 133 | if (!(all_threads_mask & (all_threads_mask - 1))) |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 134 | return; |
| 135 | |
| 136 | thread_sync_barrier(&barrier); |
| 137 | if (threads_want_sync & tid_bit) |
Christopher Faulet | 2a944ee | 2017-11-07 10:42:54 +0100 | [diff] [blame] | 138 | HA_SPIN_LOCK(THREAD_SYNC_LOCK, &sync_lock); |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 139 | } |
| 140 | |
| 141 | /* Exit from the sync point and unlock it if it was previously locked. If the |
| 142 | * current thread is the last one to have requested a sync, the sync pipe is |
| 143 | * flushed. |
| 144 | */ |
| 145 | void thread_exit_sync() |
| 146 | { |
| 147 | static volatile unsigned long barrier = 0; |
| 148 | |
Willy Tarreau | 0c026f4 | 2018-08-01 19:12:20 +0200 | [diff] [blame] | 149 | if (!(all_threads_mask & (all_threads_mask - 1))) |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 150 | return; |
| 151 | |
| 152 | if (threads_want_sync & tid_bit) |
Christopher Faulet | 2a944ee | 2017-11-07 10:42:54 +0100 | [diff] [blame] | 153 | HA_SPIN_UNLOCK(THREAD_SYNC_LOCK, &sync_lock); |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 154 | |
| 155 | if (HA_ATOMIC_AND(&threads_want_sync, ~tid_bit) == 0) { |
| 156 | char c; |
| 157 | |
| 158 | shut_your_big_mouth_gcc(read(threads_sync_pipe[0], &c, 1)); |
| 159 | fd_done_recv(threads_sync_pipe[0]); |
| 160 | } |
| 161 | |
| 162 | thread_sync_barrier(&barrier); |
| 163 | } |
| 164 | |
Willy Tarreau | 60b639c | 2018-08-02 10:16:17 +0200 | [diff] [blame] | 165 | /* Marks the thread as harmless until the last thread using the rendez-vous |
| 166 | * point quits. Given that we can wait for a long time, sched_yield() is used |
| 167 | * when available to offer the CPU resources to competing threads if needed. |
| 168 | */ |
| 169 | void thread_harmless_till_end() |
| 170 | { |
| 171 | HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); |
| 172 | while (threads_want_rdv_mask & all_threads_mask) { |
| 173 | #if _POSIX_PRIORITY_SCHEDULING |
| 174 | sched_yield(); |
| 175 | #else |
| 176 | pl_cpu_relax(); |
| 177 | #endif |
| 178 | } |
| 179 | } |
| 180 | |
| 181 | /* Isolates the current thread : request the ability to work while all other |
| 182 | * threads are harmless. Only returns once all of them are harmless, with the |
| 183 | * current thread's bit in threads_harmless_mask cleared. Needs to be completed |
| 184 | * using thread_release(). |
| 185 | */ |
| 186 | void thread_isolate() |
| 187 | { |
| 188 | unsigned long old; |
| 189 | |
| 190 | HA_ATOMIC_OR(&threads_harmless_mask, tid_bit); |
| 191 | __ha_barrier_store(); |
| 192 | HA_ATOMIC_OR(&threads_want_rdv_mask, tid_bit); |
| 193 | |
| 194 | /* wait for all threads to become harmless */ |
| 195 | old = threads_harmless_mask; |
| 196 | while (1) { |
| 197 | if (unlikely((old & all_threads_mask) != all_threads_mask)) |
| 198 | old = threads_harmless_mask; |
| 199 | else if (HA_ATOMIC_CAS(&threads_harmless_mask, &old, old & ~tid_bit)) |
| 200 | break; |
| 201 | |
| 202 | #if _POSIX_PRIORITY_SCHEDULING |
| 203 | sched_yield(); |
| 204 | #else |
| 205 | pl_cpu_relax(); |
| 206 | #endif |
| 207 | } |
| 208 | /* one thread gets released at a time here, with its harmess bit off. |
| 209 | * The loss of this bit makes the other one continue to spin while the |
| 210 | * thread is working alone. |
| 211 | */ |
| 212 | } |
| 213 | |
| 214 | /* Cancels the effect of thread_isolate() by releasing the current thread's bit |
| 215 | * in threads_want_rdv_mask and by marking this thread as harmless until the |
| 216 | * last worker finishes. |
| 217 | */ |
| 218 | void thread_release() |
| 219 | { |
| 220 | while (1) { |
| 221 | HA_ATOMIC_AND(&threads_want_rdv_mask, ~tid_bit); |
| 222 | if (!(threads_want_rdv_mask & all_threads_mask)) |
| 223 | break; |
| 224 | thread_harmless_till_end(); |
| 225 | } |
| 226 | } |
Christopher Faulet | 339fff8 | 2017-10-19 11:59:15 +0200 | [diff] [blame] | 227 | |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 228 | __attribute__((constructor)) |
| 229 | static void __hathreads_init(void) |
| 230 | { |
Christopher Faulet | 2a944ee | 2017-11-07 10:42:54 +0100 | [diff] [blame] | 231 | HA_SPIN_INIT(&sync_lock); |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 232 | #if defined(DEBUG_THREAD) || defined(DEBUG_FULL) |
| 233 | memset(lock_stats, 0, sizeof(lock_stats)); |
| 234 | #endif |
Willy Tarreau | 6dbd3e9 | 2017-11-05 11:50:18 +0100 | [diff] [blame] | 235 | hap_register_build_opts("Built with multi-threading support.", 0); |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 236 | } |
| 237 | |
Willy Tarreau | 0ccd322 | 2018-07-30 10:34:35 +0200 | [diff] [blame] | 238 | #endif // USE_THREAD |
| 239 | |
| 240 | |
| 241 | /* Parse the number of threads in argument <arg>, returns it and adjusts a few |
| 242 | * internal variables accordingly, or fails and returns zero with an error |
| 243 | * reason in <errmsg>. May be called multiple times while parsing. |
| 244 | */ |
| 245 | int parse_nbthread(const char *arg, char **err) |
| 246 | { |
| 247 | long nbthread; |
| 248 | char *errptr; |
| 249 | |
| 250 | nbthread = strtol(arg, &errptr, 10); |
| 251 | if (!*arg || *errptr) { |
| 252 | memprintf(err, "passed a missing or unparsable integer value in '%s'", arg); |
| 253 | return 0; |
| 254 | } |
| 255 | |
| 256 | #ifndef USE_THREAD |
| 257 | if (nbthread != 1) { |
| 258 | memprintf(err, "specified with a value other than 1 while HAProxy is not compiled with threads support. Please check build options for USE_THREAD"); |
| 259 | return 0; |
| 260 | } |
| 261 | #else |
| 262 | if (nbthread < 1 || nbthread > MAX_THREADS) { |
| 263 | memprintf(err, "value must be between 1 and %d (was %ld)", MAX_THREADS, nbthread); |
| 264 | return 0; |
| 265 | } |
| 266 | |
| 267 | /* we proceed like this to be sure never to overflow the left shift */ |
| 268 | all_threads_mask = 1UL << (nbthread - 1); |
| 269 | all_threads_mask |= all_threads_mask - 1; |
Christopher Faulet | 1a2b56e | 2017-10-12 16:09:09 +0200 | [diff] [blame] | 270 | #endif |
Willy Tarreau | 0ccd322 | 2018-07-30 10:34:35 +0200 | [diff] [blame] | 271 | return nbthread; |
| 272 | } |