/*
 * Copyright (C) 2012-2020 all contributors <cmogstored-public@yhbt.net>
 * License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
 */
#include "cmogstored.h"

static __thread unsigned mog_do_quit;
struct mog_thr_start_arg {
	struct mog_thrpool *tp;
	pthread_mutex_t mtx;
	pthread_cond_t cond;
	unsigned *do_quit;
};

static sigset_t quitset;

__attribute__((constructor)) static void thrpool_init(void)
{
	CHECK(int, 0, sigfillset(&quitset));
	CHECK(int, 0, sigdelset(&quitset, SIGURG));
}

/* child thread notifies the parent about its readiness */
static void *thr_start_wrapper(void *ptr)
{
	struct mog_thr_start_arg *arg = ptr;
	struct mog_thrpool *tp;

	mog_do_quit = 0;
	CHECK(int, 0, pthread_sigmask(SIG_SETMASK, &quitset, NULL));
	CHECK(int, 0, pthread_mutex_lock(&arg->mtx));

	arg->do_quit = &mog_do_quit;
	tp = arg->tp; /* arg becomes invalid once we unlock */

	CHECK(int, 0, pthread_cond_signal(&arg->cond));
	CHECK(int, 0, pthread_mutex_unlock(&arg->mtx));

	return tp->start_fn(tp->start_arg);
}

/* child thread tests if its quit flag is set and exits if it is */
void mog_thr_test_quit(void)
{
	if (__sync_add_and_fetch(&mog_do_quit, 0) != 0) {
		mog_alloc_quit();
		pthread_exit(NULL);
	}
}

bool mog_thr_prepare_quit(void)
{
	/* no barriers or atomic instructions, this is just a hint */
	return !!mog_do_quit;
}

/*
 * we no longer rely on pthreads cancellation, so our explicit checks for
 * thread quitting requires us to continuously signal a thread for death
 * in case it enters a sleeping syscall (epoll_wait/kevent) immediately
 * after checking the mog_do_quit TLS variable
 */
static void poke(pthread_t thr, int sig)
{
	int err;

	/*
	 * This is an uncommon code path and only triggered when
	 * we lower thread counts or shut down
	 */
	while ((err = pthread_kill(thr, sig)) == 0) {
		/*
		 * sleep for 10 ms, sched_yield still burns too much CPU
		 * on FreeBSD (and likely other OSes) if a thread is waiting
		 * on disk I/O.
		 */
		struct timespec ts = { .tv_sec = 0, .tv_nsec = 10e6 };
		int rc = nanosleep(&ts, NULL);

		if (rc != 0)
			assert(errno != EINVAL && "bug in using nanosleep");
	}
	assert(err == ESRCH && "pthread_kill() usage bug");
}

static bool
thr_create_fail_retry(struct mog_thrpool *tp, unsigned size,
                      unsigned long *nr_eagain, int err)
{
	/* do not leave the pool w/o threads at all */
	if (tp->n_threads == 0) {
		if ((++*nr_eagain % 1024) == 0) {
			errno = err;
			syslog(LOG_ERR, "pthread_create: %m (tries: %lu)",
			       *nr_eagain);
		}
		mog_yield();
		return true;
	} else {
		errno = err;
		syslog(LOG_ERR,
		       "pthread_create: %m, only running %u of %u threads",
		       tp->n_threads, size);
		return false;
	}
}

static bool
thrpool_add(struct mog_thrpool *tp, unsigned size, unsigned long *nr_eagain)
{
	struct mog_thr_start_arg arg = {
		.mtx = PTHREAD_MUTEX_INITIALIZER,
		.cond = PTHREAD_COND_INITIALIZER,
	};
	pthread_t *thr;
	pthread_attr_t attr;
	size_t bytes = (tp->n_threads + 1) * sizeof(struct mog_thread);
	int rc;

	assert(tp && "tp no defined");
	arg.tp = tp;
	tp->threads = xrealloc(tp->threads, bytes);

	CHECK(int, 0, pthread_attr_init(&attr));

	thr = &tp->threads[tp->n_threads].thr;

	CHECK(int, 0, pthread_mutex_lock(&arg.mtx));
	rc = pthread_create(thr, &attr, thr_start_wrapper, &arg);
	CHECK(int, 0, pthread_attr_destroy(&attr));
	if (rc == 0) {
		CHECK(int, 0, pthread_cond_wait(&arg.cond, &arg.mtx));
		tp->threads[tp->n_threads].do_quit = arg.do_quit;
	}
	CHECK(int, 0, pthread_mutex_unlock(&arg.mtx));

	if (rc == 0) {
		tp->n_threads++;
		*nr_eagain = 0;
	} else if (mog_pthread_create_retryable(rc)) {
		if (!thr_create_fail_retry(tp, size, nr_eagain, rc))
			return false;
	} else {
		assert(rc == 0 && "pthread_create usage error");
	}
	return true;
}

void mog_thrpool_set_size(struct mog_thrpool *tp, unsigned size)
{
	unsigned long nr_eagain = 0;

	CHECK(int, 0, pthread_mutex_lock(&tp->lock));

	while (size > tp->n_threads && thrpool_add(tp, size, &nr_eagain))
		/* nothing */;

	if (tp->n_threads > size) {
		unsigned i;
		int err;

		/* set the do_quit flag for all threads we kill */
		for (i = size; i < tp->n_threads; i++) {
			__sync_add_and_fetch(tp->threads[i].do_quit, 1);
			err = pthread_kill(tp->threads[i].thr, SIGURG);

			switch (err) {
			case 0:
			case ESRCH:
				break;
			default:
				assert(0 && "pthread_kill usage bug" && err);
			}
		}

		/* keep poking them to kick them out out epoll_wait/kevent */
		for (i = size; i < tp->n_threads; i++) {
			poke(tp->threads[i].thr, SIGURG);

			CHECK(int, 0, pthread_join(tp->threads[i].thr, NULL));
		}
		tp->n_threads = size;
	}
	CHECK(int, 0, pthread_mutex_unlock(&tp->lock));
}

void
mog_thrpool_start(struct mog_thrpool *tp, unsigned nthr,
                  void *(*start_fn)(void *), void *arg)
{
	/* we may be started on a new server before device dirs exist */
	if (nthr == 0)
		nthr = 1;

	tp->threads = NULL;
	tp->n_threads = 0;
	tp->start_fn = start_fn;
	tp->start_arg = arg;
	CHECK(int, 0, pthread_mutex_init(&tp->lock, NULL));
	mog_thrpool_set_size(tp, nthr);
}

void mog_thrpool_quit(struct mog_thrpool *tp, struct mog_queue *q)
{
	mog_thrpool_set_size(tp, 0);
	CHECK(int, 0, pthread_mutex_destroy(&tp->lock));
	mog_free_and_null(&tp->threads);
}