// SPDX-License-Identifier: GPL-2.0-or-later

/* PASST - Plug A Simple Socket Transport
 *  for qemu/UNIX domain socket mode
 *
 * PASTA - Pack A Subtle Tap Abstraction
 *  for network namespace/tap device mode
 *
 * threading.c - Basic threading infrastructure
 *
 * Copyright Red Hat
 * Author: Laurent Vivier <lvivier@redhat.com>
 */

#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>

#include "util.h"
#include "passt.h"
#include "threading.h"

/**
 * struct threading_context - Per-thread execution context
 * @epollfd:	Epoll file descriptor for this thread
 * @pthread:	Pthread identifier
 * @worker:	Worker function to execute in this thread
 * @opaque:	Opaque data passed to worker function
 * @quit:	Flag to signal thread to exit
 */
struct threading_context {
	pthread_t pthread;
	void (*worker)(void *, int, struct epoll_event *);
	void *opaque;
	int epollfd;
	_Atomic bool quit;
};

/*
 *  index 0 is reserved for main process
 *  new threads start at index 1
 */
struct threading_context threads[MAX_NUM_THREADS + 1];

/**
 * threading_init() - Initialize threading infrastructure
 */
void threading_init(void)
{
	int i;

	for (i = 0; i < ARRAY_SIZE(threads); i++) {
		threads[i].epollfd = epoll_create1(EPOLL_CLOEXEC);
		if (threads[i].epollfd == -1)
			die_perror("Failed to create epoll for threads #%d",
				   i);
	}
}

/**
 * threading_worker_set() - Set worker function for a thread
 * @threadid:		Thread index
 * @worker:		Worker function to assign to the thread
 * @is_valid:	Validation function to select a thread
 * @opaque:		Opaque data to pass to worker function
 *
 * Return: 0 on success, -1 if thread index is invalid
 */
/* cppcheck-suppress unusedFunction */
int threading_worker_set(unsigned int threadid,
			 void (*worker)(void *, int, struct epoll_event *),
			 void *opaque)
{
	struct threading_context *tc;

	if (threadid >= ARRAY_SIZE(threads))
		return -1;

	tc = &threads[threadid];

	tc->worker = worker;
	tc->opaque = opaque;

	return 0;
}

/**
 * threading_epollfd() - Get epoll file descriptor for a thread
 * @threadid:	Thread index
 *
 * Return: epoll file descriptor for the specified thread, -1 if index invalid
 */
/* cppcheck-suppress unusedFunction */
int threading_epollfd(unsigned int threadid)
{
	if (threadid >= ARRAY_SIZE(threads))
		return -1;

	return threads[threadid].epollfd;
}

/**
 * threading_worker() - Main worker thread function
 * @opaque:	Pointer to threading_context for this thread
 *
 * Return: NULL on thread exit
 *
 * #syscalls poll write futex
 */
static void *threading_worker(void *opaque)
{
	struct threading_context *tc = opaque;

	while (!tc->quit) {
		struct epoll_event events[NUM_EPOLL_EVENTS];
		int nfds;

		/* NOLINTBEGIN(bugprone-branch-clone): intervals can be the same */
		/* cppcheck-suppress [duplicateValueTernary, unmatchedSuppression] */
		nfds = epoll_wait(tc->epollfd, events, NUM_EPOLL_EVENTS,
				  TIMER_INTERVAL);
		/* NOLINTEND(bugprone-branch-clone) */
		if (nfds == -1 && errno != EINTR)
			die_perror("epoll_wait() failed in thread loop");

		tc->worker(tc->opaque, nfds, events);
	}

	return NULL;
}

/**
 * threading_start_thread() - Start a worker thread
 * @threadid:	Thread index to start
 *
 * #syscalls rt_sigaction rt_sigprocmask mprotect getrandom brk clone3 rseq set_robust_list clock_nanosleep
 */
/* cppcheck-suppress unusedFunction */
void threading_start_thread(unsigned int threadid)
{
	struct threading_context *tc;
	int ret;

	if (threadid >= ARRAY_SIZE(threads))
		die_perror("Invalid thread index %u, max is %u\n",
			   threadid, MAX_NUM_THREADS);

	tc = &threads[threadid];

	if (!tc->worker)
		die_perror("No worker for thread #%u\n", threadid);

	if (tc->pthread)
		return;

	/* thread #0 is the main process */
	if (threadid == THREADING_ID_DEFAULT) {
		tc->pthread = -1; /* Mark thread is active */
		tc->quit = false;
		/* Thread #0 is main process, call is blocking */
		threading_worker(tc);	/* no return */
	}

	tc->quit = false;
	ret = pthread_create(&tc->pthread, NULL, &threading_worker, tc);
	if (ret != 0)
		die_perror("pthread_create() failed: thread index %u ret %d",
			   threadid, ret);
}

/**
 * threading_stop_thread() - Stop and join a worker thread
 * @threadid:	Thread index to stop
 */
/* cppcheck-suppress unusedFunction */
void threading_stop_thread(unsigned int threadid)
{
	if (threadid >= ARRAY_SIZE(threads))
		die_perror("Invalid thread index %u, max is %u\n",
		           threadid, ARRAY_SIZE(threads));

	if (threadid == THREADING_ID_DEFAULT) {
		/* Thread #0 is main process, cannot be stopped */
		return;
	}

	threads[threadid].quit = true;
	pthread_join(threads[threadid].pthread, NULL);
	threads[threadid].pthread = 0;
}

/**
 * threading_is_active() - Check if a worker thread is active
 * @threadid:	Thread index to check
 *
 * Return: true if the thread at the given index has been created and is active,
 *         false otherwise
 */
/* cppcheck-suppress unusedFunction */
bool threading_is_active(unsigned int threadid)
{
	if (threadid >= ARRAY_SIZE(threads))
		die_perror("Invalid thread index %u, max is %u\n",
		           threadid, ARRAY_SIZE(threads));

	return !!threads[threadid].pthread;
}