lib/frr_pthread.c

   1 /*
   2  * Utilities and interfaces for managing POSIX threads within FRR.
   3  * Copyright (C) 2017  Cumulus Networks, Inc.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; see the file COPYING; if not, write to the Free Software
  17  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  18  */
  19
  20 #include <zebra.h>
  21 #include <pthread.h>
  22 #include <sched.h>
  23
  24 #include "frr_pthread.h"
  25 #include "memory.h"
  26 #include "hash.h"
  27
  28 DEFINE_MTYPE(LIB, FRR_PTHREAD, "FRR POSIX Thread");
  29 DEFINE_MTYPE(LIB, PTHREAD_PRIM, "POSIX synchronization primitives");
  30
  31 /* id for next created pthread */
  32 static _Atomic uint32_t next_id = 0;
  33
  34 /* default frr_pthread start/stop routine prototypes */
  35 static void *fpt_run(void *arg);
  36 static int fpt_halt(struct frr_pthread *fpt, void **res);
  37
  38 /* default frr_pthread attributes */
  39 struct frr_pthread_attr frr_pthread_attr_default = {
  40         .id = 0,
  41         .start = fpt_run,
  42         .stop = fpt_halt,
  43 };
  44
  45 /* hash table to keep track of all frr_pthreads */
  46 static struct hash *frr_pthread_hash;
  47 static pthread_mutex_t frr_pthread_hash_mtx = PTHREAD_MUTEX_INITIALIZER;
  48
  49 /* frr_pthread_hash->hash_cmp */
  50 static int frr_pthread_hash_cmp(const void *value1, const void *value2)
  51 {
  52         const struct frr_pthread *tq1 = value1;
  53         const struct frr_pthread *tq2 = value2;
  54
  55         return (tq1->attr.id == tq2->attr.id);
  56 }
  57
  58 /* frr_pthread_hash->hash_key */
  59 static unsigned int frr_pthread_hash_key(void *value)
  60 {
  61         return ((struct frr_pthread *)value)->attr.id;
  62 }
  63
  64 /* ------------------------------------------------------------------------ */
  65
  66 void frr_pthread_init()
  67 {
  68         pthread_mutex_lock(&frr_pthread_hash_mtx);
  69         {
  70                 frr_pthread_hash = hash_create(frr_pthread_hash_key,
  71                                                frr_pthread_hash_cmp, NULL);
  72         }
  73         pthread_mutex_unlock(&frr_pthread_hash_mtx);
  74 }
  75
  76 void frr_pthread_finish()
  77 {
  78         pthread_mutex_lock(&frr_pthread_hash_mtx);
  79         {
  80                 hash_clean(frr_pthread_hash,
  81                            (void (*)(void *))frr_pthread_destroy);
  82                 hash_free(frr_pthread_hash);
  83         }
  84         pthread_mutex_unlock(&frr_pthread_hash_mtx);
  85 }
  86
  87 struct frr_pthread *frr_pthread_new(struct frr_pthread_attr *attr,
  88                                     const char *name, const char *os_name)
  89 {
  90         static struct frr_pthread holder = {};
  91         struct frr_pthread *fpt = NULL;
  92
  93         attr = attr ? attr : &frr_pthread_attr_default;
  94
  95         pthread_mutex_lock(&frr_pthread_hash_mtx);
  96         {
  97                 holder.attr.id = attr->id;
  98
  99                 if (!hash_lookup(frr_pthread_hash, &holder)) {
 100                         fpt = XCALLOC(MTYPE_FRR_PTHREAD,
 101                                       sizeof(struct frr_pthread));
 102                         /* initialize mutex */
 103                         pthread_mutex_init(&fpt->mtx, NULL);
 104                         /* create new thread master */
 105                         fpt->master = thread_master_create(name);
 106                         /* set attributes */
 107                         fpt->attr = *attr;
 108                         name = (name ? name : "Anonymous thread");
 109                         fpt->name = XSTRDUP(MTYPE_FRR_PTHREAD, name);
 110                         if (os_name)
 111                                 snprintf(fpt->os_name, OS_THREAD_NAMELEN,
 112                                          "%s", os_name);
 113                         if (attr == &frr_pthread_attr_default)
 114                                 fpt->attr.id = frr_pthread_get_id();
 115                         /* initialize startup synchronization primitives */
 116                         fpt->running_cond_mtx = XCALLOC(
 117                                 MTYPE_PTHREAD_PRIM, sizeof(pthread_mutex_t));
 118                         fpt->running_cond = XCALLOC(MTYPE_PTHREAD_PRIM,
 119                                                     sizeof(pthread_cond_t));
 120                         pthread_mutex_init(fpt->running_cond_mtx, NULL);
 121                         pthread_cond_init(fpt->running_cond, NULL);
 122
 123                         /* insert into global thread hash */
 124                         hash_get(frr_pthread_hash, fpt, hash_alloc_intern);
 125                 }
 126         }
 127         pthread_mutex_unlock(&frr_pthread_hash_mtx);
 128
 129         return fpt;
 130 }
 131
 132 void frr_pthread_destroy(struct frr_pthread *fpt)
 133 {
 134         thread_master_free(fpt->master);
 135
 136         pthread_mutex_destroy(&fpt->mtx);
 137         pthread_mutex_destroy(fpt->running_cond_mtx);
 138         pthread_cond_destroy(fpt->running_cond);
 139         if (fpt->name)
 140                 XFREE(MTYPE_FRR_PTHREAD, fpt->name);
 141         XFREE(MTYPE_PTHREAD_PRIM, fpt->running_cond_mtx);
 142         XFREE(MTYPE_PTHREAD_PRIM, fpt->running_cond);
 143         XFREE(MTYPE_FRR_PTHREAD, fpt);
 144 }
 145
 146 int frr_pthread_set_name(struct frr_pthread *fpt, const char *name,
 147                          const char *os_name)
 148 {
 149         int ret = 0;
 150
 151         if (name) {
 152                 pthread_mutex_lock(&fpt->mtx);
 153                 {
 154                         if (fpt->name)
 155                                 XFREE(MTYPE_FRR_PTHREAD, fpt->name);
 156                         fpt->name = XSTRDUP(MTYPE_FRR_PTHREAD, name);
 157                 }
 158                 pthread_mutex_unlock(&fpt->mtx);
 159                 thread_master_set_name(fpt->master, name);
 160         }
 161
 162         if (os_name) {
 163                 pthread_mutex_lock(&fpt->mtx);
 164                 snprintf(fpt->os_name, OS_THREAD_NAMELEN, "%s", os_name);
 165                 pthread_mutex_unlock(&fpt->mtx);
 166 #ifdef GNU_LINUX
 167                 ret = pthread_setname_np(fpt->thread, fpt->os_name);
 168 #elif defined(OPEN_BSD)
 169                 ret = pthread_set_name_np(fpt->thread, fpt->os_name);
 170 #endif
 171         }
 172
 173         return ret;
 174 }
 175
 176 struct frr_pthread *frr_pthread_get(uint32_t id)
 177 {
 178         static struct frr_pthread holder = {};
 179         struct frr_pthread *fpt;
 180
 181         pthread_mutex_lock(&frr_pthread_hash_mtx);
 182         {
 183                 holder.attr.id = id;
 184                 fpt = hash_lookup(frr_pthread_hash, &holder);
 185         }
 186         pthread_mutex_unlock(&frr_pthread_hash_mtx);
 187
 188         return fpt;
 189 }
 190
 191 int frr_pthread_run(struct frr_pthread *fpt, const pthread_attr_t *attr)
 192 {
 193         int ret;
 194
 195         ret = pthread_create(&fpt->thread, attr, fpt->attr.start, fpt);
 196
 197         /*
 198          * Per pthread_create(3), the contents of fpt->thread are undefined if
 199          * pthread_create() did not succeed. Reset this value to zero.
 200          */
 201         if (ret < 0)
 202                 memset(&fpt->thread, 0x00, sizeof(fpt->thread));
 203
 204         return ret;
 205 }
 206
 207 void frr_pthread_wait_running(struct frr_pthread *fpt)
 208 {
 209         pthread_mutex_lock(fpt->running_cond_mtx);
 210         {
 211                 while (!fpt->running)
 212                         pthread_cond_wait(fpt->running_cond,
 213                                           fpt->running_cond_mtx);
 214         }
 215         pthread_mutex_unlock(fpt->running_cond_mtx);
 216 }
 217
 218 void frr_pthread_notify_running(struct frr_pthread *fpt)
 219 {
 220         pthread_mutex_lock(fpt->running_cond_mtx);
 221         {
 222                 fpt->running = true;
 223                 pthread_cond_signal(fpt->running_cond);
 224         }
 225         pthread_mutex_unlock(fpt->running_cond_mtx);
 226 }
 227
 228 int frr_pthread_stop(struct frr_pthread *fpt, void **result)
 229 {
 230         int ret = (*fpt->attr.stop)(fpt, result);
 231         memset(&fpt->thread, 0x00, sizeof(fpt->thread));
 232         return ret;
 233 }
 234
 235 /*
 236  * Callback for hash_iterate to stop all frr_pthread's.
 237  */
 238 static void frr_pthread_stop_all_iter(struct hash_backet *hb, void *arg)
 239 {
 240         struct frr_pthread *fpt = hb->data;
 241         frr_pthread_stop(fpt, NULL);
 242 }
 243
 244 void frr_pthread_stop_all()
 245 {
 246         pthread_mutex_lock(&frr_pthread_hash_mtx);
 247         {
 248                 hash_iterate(frr_pthread_hash, frr_pthread_stop_all_iter, NULL);
 249         }
 250         pthread_mutex_unlock(&frr_pthread_hash_mtx);
 251 }
 252
 253 uint32_t frr_pthread_get_id(void)
 254 {
 255         _Atomic uint32_t nxid;
 256         nxid = atomic_fetch_add_explicit(&next_id, 1, memory_order_seq_cst);
 257         /* just a sanity check, this should never happen */
 258         assert(nxid <= (UINT32_MAX - 1));
 259         return nxid;
 260 }
 261
 262 void frr_pthread_yield(void)
 263 {
 264         (void)sched_yield();
 265 }
 266
 267 /*
 268  * ----------------------------------------------------------------------------
 269  * Default Event Loop
 270  * ----------------------------------------------------------------------------
 271  */
 272
 273 /* dummy task for sleeper pipe */
 274 static int fpt_dummy(struct thread *thread)
 275 {
 276         return 0;
 277 }
 278
 279 /* poison pill task to end event loop */
 280 static int fpt_finish(struct thread *thread)
 281 {
 282         struct frr_pthread *fpt = THREAD_ARG(thread);
 283
 284         atomic_store_explicit(&fpt->running, false, memory_order_relaxed);
 285         return 0;
 286 }
 287
 288 /* stop function, called from other threads to halt this one */
 289 static int fpt_halt(struct frr_pthread *fpt, void **res)
 290 {
 291         thread_add_event(fpt->master, &fpt_finish, fpt, 0, NULL);
 292         pthread_join(fpt->thread, res);
 293
 294         return 0;
 295 }
 296
 297 /*
 298  * Entry pthread function & main event loop.
 299  *
 300  * Upon thread start the following actions occur:
 301  *
 302  * - frr_pthread's owner field is set to pthread ID.
 303  * - All signals are blocked (except for unblockable signals).
 304  * - Pthread's threadmaster is set to never handle pending signals
 305  * - Poker pipe for poll() is created and queued as I/O source
 306  * - The frr_pthread->running_cond condition variable is signalled to indicate
 307  *   that the previous actions have completed. It is not safe to assume any of
 308  *   the above have occurred before receiving this signal.
 309  *
 310  * After initialization is completed, the event loop begins running. Each tick,
 311  * the following actions are performed before running the usual event system
 312  * tick function:
 313  *
 314  * - Verify that the running boolean is set
 315  * - Verify that there are no pending cancellation requests
 316  * - Verify that there are tasks scheduled
 317  *
 318  * So long as the conditions are met, the event loop tick is run and the
 319  * returned task is executed.
 320  *
 321  * If any of these conditions are not met, the event loop exits, closes the
 322  * pipes and dies without running any cleanup functions.
 323  */
 324 static void *fpt_run(void *arg)
 325 {
 326         struct frr_pthread *fpt = arg;
 327         fpt->master->owner = pthread_self();
 328
 329         int sleeper[2];
 330         pipe(sleeper);
 331         thread_add_read(fpt->master, &fpt_dummy, NULL, sleeper[0], NULL);
 332
 333         fpt->master->handle_signals = false;
 334
 335         if (fpt->os_name[0])
 336                 frr_pthread_set_name(fpt, NULL, fpt->os_name);
 337
 338         frr_pthread_notify_running(fpt);
 339
 340         struct thread task;
 341         while (atomic_load_explicit(&fpt->running, memory_order_relaxed)) {
 342                 pthread_testcancel();
 343                 if (thread_fetch(fpt->master, &task)) {
 344                         thread_call(&task);
 345                 }
 346         }
 347
 348         close(sleeper[1]);
 349         close(sleeper[0]);
 350
 351         return NULL;
 352 }