// SPDX-License-Identifier: ISC

/*

 * Copyright (c) 2017-19  David Lamparter, for NetDEF, Inc.

 */

/* implementation notes:  this is an epoch-based RCU implementation.  rcu_seq

 * (global variable) counts the current epoch.  Threads hold a specific epoch

 * in rcu_read_lock().  This is the oldest epoch a thread might be accessing

 * data from.

 *

 * The rcu_seq global is only pushed forward on rcu_read_lock() and

 * rcu_read_unlock() calls.  This makes things a tad more efficient since

 * those are the only places it matters:

 * - on rcu_read_lock, we don't want to hold an old epoch pointlessly

 * - on rcu_read_unlock, we want to make sure we're not stuck on an old epoch

 *   when heading into a long idle period where no thread holds RCU

 *

 * rcu_thread structures themselves are RCU-free'd.

 *

 * rcu_head structures are the most iffy;  normally for an ATOMLIST we would

 * need to make sure we use rcu_free or pthread_rwlock to deallocate old items

 * to prevent ABA or use-after-free problems.  However, our ATOMLIST code

 * guarantees that if the list remains non-empty in all cases, we only need

 * the "last" pointer to do an "add_tail()", i.e. we can't run into ABA/UAF

 * issues - but we do need to keep at least 1 item on the list.

 *

 * (Search the atomlist code for all uses of "last")

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include <pthread.h>

#ifdef HAVE_PTHREAD_NP_H

#include <pthread_np.h>

#endif

#include <string.h>

#include <unistd.h>

#include <signal.h>

#include "frrcu.h"

#include "seqlock.h"

#include "atomlist.h"

DEFINE_MTYPE_STATIC(LIB, RCU_THREAD,    "RCU thread");

DEFINE_MTYPE_STATIC(LIB, RCU_NEXT,      "RCU sequence barrier");

DECLARE_ATOMLIST(rcu_heads, struct rcu_head, head);

PREDECL_ATOMLIST(rcu_threads);

struct rcu_thread {

  struct rcu_threads_item head;

  struct rcu_head rcu_head;

  struct seqlock rcu;

  /* only accessed by thread itself, not atomic */

  unsigned depth;

};

DECLARE_ATOMLIST(rcu_threads, struct rcu_thread, head);

static const struct rcu_action rcua_next  = { .type = RCUA_NEXT };

static const struct rcu_action rcua_end   = { .type = RCUA_END };

static const struct rcu_action rcua_close = { .type = RCUA_CLOSE };

struct rcu_next {

  struct rcu_head head_free;

  struct rcu_head head_next;

};

#define rcu_free_internal(mtype, ptr, field)                                   \

  do {                                                                   \

    typeof(ptr) _ptr = (ptr);                                      \

    struct rcu_head *_rcu_head = &_ptr->field;                     \

    static const struct rcu_action _rcu_action = {                 \

      .type = RCUA_FREE,                                     \

      .u.free = {                                            \

        .mt = mtype,                                   \

        .offset = offsetof(typeof(*_ptr), field),      \

      },                                                     \

    };                                                             \

    _rcu_head->action = &_rcu_action;                              \

    rcu_heads_add_tail(&rcu_heads, _rcu_head);                     \

  } while (0)

/* primary global RCU position */

static struct seqlock rcu_seq;

/* this is set to rcu_seq whenever something is added on the RCU queue.

 * rcu_read_lock() and rcu_read_unlock() will then bump rcu_seq up one step.

 */

static _Atomic seqlock_val_t rcu_dirty;

static struct rcu_threads_head rcu_threads;

static struct rcu_heads_head rcu_heads;

/* main thread & RCU sweeper have pre-setup rcu_thread structures.  The

 * reasons are different:

 *

 * - rcu_thread_main is there because the main thread isn't started like

 *   other threads, it's implicitly created when the program is started.  So

 *   rcu_thread_main matches up implicitly.

 *

 * - rcu_thread_rcu isn't actually put on the rcu_threads list (makes no

 *   sense really), it only exists so we can call RCU-using functions from

 *   the RCU thread without special handling in rcu_read_lock/unlock.

 */

static struct rcu_thread rcu_thread_main;

static struct rcu_thread rcu_thread_rcu;

static pthread_t rcu_pthread;

static pthread_key_t rcu_thread_key;

static bool rcu_active;

static void rcu_start(void);

static void rcu_bump(void);

/*

 * preinitialization for main thread

 */

static void rcu_thread_end(void *rcu_thread);

static void rcu_preinit(void) __attribute__((constructor));

static void rcu_preinit(void)

{

  struct rcu_thread *rt;

  rt = &rcu_thread_main;

  rt->depth = 1;

  seqlock_init(&rt->rcu);

  seqlock_acquire_val(&rt->rcu, SEQLOCK_STARTVAL);

  pthread_key_create(&rcu_thread_key, rcu_thread_end);

  pthread_setspecific(rcu_thread_key, rt);

  rcu_threads_add_tail(&rcu_threads, rt);

  /* RCU sweeper's rcu_thread is a dummy, NOT added to rcu_threads */

  rt = &rcu_thread_rcu;

  rt->depth = 1;

  seqlock_init(&rcu_seq);

  seqlock_acquire_val(&rcu_seq, SEQLOCK_STARTVAL);

}

static struct rcu_thread *rcu_self(void)

{

  return (struct rcu_thread *)pthread_getspecific(rcu_thread_key);

}

/*

 * thread management (for the non-main thread)

 */

struct rcu_thread *rcu_thread_prepare(void)

{

  struct rcu_thread *rt, *cur;

  rcu_assert_read_locked();

  if (!rcu_active)

    rcu_start();

  cur = rcu_self();

  assert(cur->depth);

  /* new thread always starts with rcu_read_lock held at depth 1, and

   * holding the same epoch as the parent (this makes it possible to

   * use RCU for things passed into the thread through its arg)

   */

  rt = XCALLOC(MTYPE_RCU_THREAD, sizeof(*rt));

  rt->depth = 1;

  seqlock_init(&rt->rcu);

  seqlock_acquire(&rt->rcu, &cur->rcu);

  rcu_threads_add_tail(&rcu_threads, rt);

  return rt;

}

void rcu_thread_start(struct rcu_thread *rt)

{

  pthread_setspecific(rcu_thread_key, rt);

}

void rcu_thread_unprepare(struct rcu_thread *rt)

{

  if (rt == &rcu_thread_rcu)

    return;

  rt->depth = 1;

  seqlock_acquire(&rt->rcu, &rcu_seq);

  rcu_bump();

  if (rt != &rcu_thread_main)

    /* this free() happens after seqlock_release() below */

    rcu_free_internal(MTYPE_RCU_THREAD, rt, rcu_head);

  rcu_threads_del(&rcu_threads, rt);

  seqlock_release(&rt->rcu);

}

static void rcu_thread_end(void *rtvoid)

{

  struct rcu_thread *rt = rtvoid;

  rcu_thread_unprepare(rt);

}

/*

 * main RCU control aspects

 */

static void rcu_bump(void)

{

  struct rcu_next *rn;

  rn = XMALLOC(MTYPE_RCU_NEXT, sizeof(*rn));

  /* note: each RCUA_NEXT item corresponds to exactly one seqno bump.

   * This means we don't need to communicate which seqno is which

   * RCUA_NEXT, since we really don't care.

   */

  /*

   * Important race condition:  while rcu_heads_add_tail is executing,

   * there is an intermediate point where the rcu_heads "last" pointer

   * already points to rn->head_next, but rn->head_next isn't added to

   * the list yet.  That means any other "add_tail" calls append to this

   * item, which isn't fully on the list yet.  Freeze this thread at

   * that point and look at another thread doing a rcu_bump.  It adds

   * these two items and then does a seqlock_bump.  But the rcu_heads

   * list is still "interrupted" and there's no RCUA_NEXT on the list

   * yet (from either the frozen thread or the second thread).  So

   * rcu_main() might actually hit the end of the list at the

   * "interrupt".

   *

   * This situation is prevented by requiring that rcu_read_lock is held

   * for any calls to rcu_bump, since if we're holding the current RCU

   * epoch, that means rcu_main can't be chewing on rcu_heads and hit

   * that interruption point.  Only by the time the thread has continued

   * to rcu_read_unlock() - and therefore completed the add_tail - the

   * RCU sweeper gobbles up the epoch and can be sure to find at least

   * the RCUA_NEXT and RCUA_FREE items on rcu_heads.

   */

  rn->head_next.action = &rcua_next;

  rcu_heads_add_tail(&rcu_heads, &rn->head_next);

  /* free rn that we allocated above.

   *

   * This is INTENTIONALLY not built into the RCUA_NEXT action.  This

   * ensures that after the action above is popped off the queue, there

   * is still at least 1 item on the RCU queue.  This means we never

   * delete the last item, which is extremely important since it keeps

   * the atomlist ->last pointer alive and well.

   *

   * If we were to "run dry" on the RCU queue, add_tail may run into the

   * "last item is being deleted - start over" case, and then we may end

   * up accessing old RCU queue items that are already free'd.

   */

  rcu_free_internal(MTYPE_RCU_NEXT, rn, head_free);

  /* Only allow the RCU sweeper to run after these 2 items are queued.

   *

   * If another thread enqueues some RCU action in the intermediate

   * window here, nothing bad happens - the queued action is associated

   * with a larger seq# than strictly necessary.  Thus, it might get

   * executed a bit later, but that's not a problem.

   *

   * If another thread acquires the read lock in this window, it holds

   * the previous epoch, but its RCU queue actions will be in the next

   * epoch.  This isn't a problem either, just a tad inefficient.

   */

  seqlock_bump(&rcu_seq);

}

static void rcu_bump_maybe(void)

{

  seqlock_val_t dirty;

  dirty = atomic_load_explicit(&rcu_dirty, memory_order_relaxed);

  /* no problem if we race here and multiple threads bump rcu_seq;

   * bumping too much causes no issues while not bumping enough will

   * result in delayed cleanup

   */

  if (dirty == seqlock_cur(&rcu_seq))

    rcu_bump();

}

void rcu_read_lock(void)

{

  struct rcu_thread *rt = rcu_self();

  assert(rt);

  if (rt->depth++ > 0)

    return;

  seqlock_acquire(&rt->rcu, &rcu_seq);

  /* need to hold RCU for bump ... */

  rcu_bump_maybe();

  /* ... but no point in holding the old epoch if we just bumped */

  seqlock_acquire(&rt->rcu, &rcu_seq);

}

void rcu_read_unlock(void)

{

  struct rcu_thread *rt = rcu_self();

  assert(rt && rt->depth);

  if (--rt->depth > 0)

    return;

  rcu_bump_maybe();

  seqlock_release(&rt->rcu);

}

void rcu_assert_read_locked(void)

{

  struct rcu_thread *rt = rcu_self();

  assert(rt && rt->depth && seqlock_held(&rt->rcu));

}

void rcu_assert_read_unlocked(void)

{

  struct rcu_thread *rt = rcu_self();

  assert(rt && !rt->depth && !seqlock_held(&rt->rcu));

}

/*

 * RCU resource-release thread

 */

static void *rcu_main(void *arg);

static void rcu_start(void)

{

  /* ensure we never handle signals on the RCU thread by blocking

   * everything here (new thread inherits signal mask)

   */

  sigset_t oldsigs, blocksigs;

  sigfillset(&blocksigs);

  pthread_sigmask(SIG_BLOCK, &blocksigs, &oldsigs);

  rcu_active = true;

  assert(!pthread_create(&rcu_pthread, NULL, rcu_main, NULL));

  pthread_sigmask(SIG_SETMASK, &oldsigs, NULL);

#ifdef HAVE_PTHREAD_SETNAME_NP

# ifdef GNU_LINUX

  pthread_setname_np(rcu_pthread, "RCU sweeper");

# elif defined(__NetBSD__)

  pthread_setname_np(rcu_pthread, "RCU sweeper", NULL);

# endif

#elif defined(HAVE_PTHREAD_SET_NAME_NP)

  pthread_set_name_np(rcu_pthread, "RCU sweeper");

#endif

}

static void rcu_do(struct rcu_head *rh)

{

  struct rcu_head_close *rhc;

  void *p;

  switch (rh->action->type) {

  case RCUA_FREE:

    p = (char *)rh - rh->action->u.free.offset;

    if (rh->action->u.free.mt)

      qfree(rh->action->u.free.mt, p);

    else

      free(p);

    break;

  case RCUA_CLOSE:

    rhc = container_of(rh, struct rcu_head_close,

           rcu_head);

    close(rhc->fd);

    break;

  case RCUA_CALL:

    p = (char *)rh - rh->action->u.call.offset;

    rh->action->u.call.fptr(p);

    break;

  case RCUA_INVALID:

  case RCUA_NEXT:

  case RCUA_END:

  default:

    assert(0);

  }

}

static void rcu_watchdog(struct rcu_thread *rt)

{

#if 0

  /* future work: print a backtrace for the thread that's holding up

   * RCU.  The only (good) way of doing that is to send a signal to the

   * other thread, save away the backtrace in the signal handler, and

   * block here until the signal is done processing.

   *

   * Just haven't implemented that yet.

   */

  fprintf(stderr, "RCU watchdog %p\n", rt);

#endif

}

static void *rcu_main(void *arg)

{

  struct rcu_thread *rt;

  struct rcu_head *rh = NULL;

  bool end = false;

  struct timespec maxwait;

  seqlock_val_t rcuval = SEQLOCK_STARTVAL;

  pthread_setspecific(rcu_thread_key, &rcu_thread_rcu);

  while (!end) {

    seqlock_wait(&rcu_seq, rcuval);

    /* RCU watchdog timeout, TODO: configurable value */

    clock_gettime(CLOCK_MONOTONIC, &maxwait);

    maxwait.tv_nsec += 100 * 1000 * 1000;

    if (maxwait.tv_nsec >= 1000000000) {

      maxwait.tv_sec++;

      maxwait.tv_nsec -= 1000000000;

    }

    frr_each (rcu_threads, &rcu_threads, rt)

      if (!seqlock_timedwait(&rt->rcu, rcuval, &maxwait)) {

        rcu_watchdog(rt);

        seqlock_wait(&rt->rcu, rcuval);

      }

    while ((rh = rcu_heads_pop(&rcu_heads))) {

      if (rh->action->type == RCUA_NEXT)

        break;

      else if (rh->action->type == RCUA_END)

        end = true;

      else

        rcu_do(rh);

    }

    rcuval += SEQLOCK_INCR;

  }

  /* rcu_shutdown can only be called singlethreaded, and it does a

   * pthread_join, so it should be impossible that anything ended up

   * on the queue after RCUA_END

   */

#if 1

  assert(!rcu_heads_first(&rcu_heads));

#else

  while ((rh = rcu_heads_pop(&rcu_heads)))

    if (rh->action->type >= RCUA_FREE)

      rcu_do(rh);

#endif

  return NULL;

}

void rcu_shutdown(void)

{

  static struct rcu_head rcu_head_end;

  struct rcu_thread *rt = rcu_self();

  void *retval;

  if (!rcu_active)

    return;

  rcu_assert_read_locked();

  assert(rcu_threads_count(&rcu_threads) == 1);

  rcu_enqueue(&rcu_head_end, &rcua_end);

  rt->depth = 0;

  seqlock_release(&rt->rcu);

  seqlock_release(&rcu_seq);

  rcu_active = false;

  /* clearing rcu_active is before pthread_join in case we hang in

   * pthread_join & get a SIGTERM or something - in that case, just

   * ignore the maybe-still-running RCU thread

   */

  if (pthread_join(rcu_pthread, &retval) == 0) {

    seqlock_acquire_val(&rcu_seq, SEQLOCK_STARTVAL);

    seqlock_acquire_val(&rt->rcu, SEQLOCK_STARTVAL);

    rt->depth = 1;

  }

}

/*

 * RCU'd free functions

 */

void rcu_enqueue(struct rcu_head *rh, const struct rcu_action *action)

{

  /* refer to rcu_bump() for why we need to hold RCU when adding items

   * to rcu_heads

   */

  rcu_assert_read_locked();

  rh->action = action;

  if (!rcu_active) {

    rcu_do(rh);

    return;

  }

  rcu_heads_add_tail(&rcu_heads, rh);

  atomic_store_explicit(&rcu_dirty, seqlock_cur(&rcu_seq),

            memory_order_relaxed);

}

void rcu_close(struct rcu_head_close *rhc, int fd)

{

  rhc->fd = fd;

  rcu_enqueue(&rhc->rcu_head, &rcua_close);

}