inetb: a single-process multithreaded inetd for pre-BONE BeOS R5

inetb.c

/* inetb: a single-process multithreaded inetd for pre-BONE BeOS R5
** Intel and PowerPC compatible
** MAKE BEBOXES GREAT AGAIN
**
** with Metrowerks:
** cc -O7 -o inetb inetb.c
** (debugging to stderr: cc -DDEBUG -o inetb inetb.c )
** (debugging to socket: cc -DSDEBUG -o inetb inetb.c )
** (you can do -DDEBUG -DSDEBUG together! it's fun!)
**
** usage: inetb port# executable [arg arg arg ...]
**
** NOTE: stderr is intentionally NOT redirected to the socket so that you
** can log and debug any issues that occur from your executable. 
**
** Copyright (C)2021 by Cameron Kaiser <ckaiser@floodgap.com>.
** Originally based on micro_inetd.
** Copyright (C)1996,2000 by Jef Poskanzer <jef@mail.acme.com>.
** All rights reserved.
**
** Redistribution and use in source and binary forms, with or without
** modification, are permitted provided that the following conditions
** are met:
** 1. Redistributions of source code must retain the above copyright
**	notice, this list of conditions and the following disclaimer.
** 2. Redistributions in binary form must reproduce the above copyright
**	notice, this list of conditions and the following disclaimer in the
**	documentation and/or other materials provided with the distribution.
**
** THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
** ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
** IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
** ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
** FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
** DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
** OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
** HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
** LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
** OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
** SUCH DAMAGE.
*/

#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <memory.h>
#include <sys/types.h>
#include <sys/param.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <scheduler.h>
#define PF_INET 2

struct iothreadstate {
	/* pid of dependent process (key, -1 if free) */
	int pid;

	/* current socket */
	int conn_fd;

	/* handles to dependent process */
	int rfd;
	int wfd;

	/* threads of i/o handlers for dependent process */
	int rtid;
	int wtid;
};
/* increase this for more simultaneous processes */
#define NUMSOCKS 32
struct iothreadstate iothreads[NUMSOCKS];
unsigned int nsock = 0;
unsigned int csock = 0;

/* on Mwerks: remember no local stack allocations > 32K */
#define BUFFERSIZE 16384

/* I'm just too lazy to inline this. it only gets called once. */
static int initialize_listen_socket(int pf, int af, unsigned short port);

static void child_handler(int sig);

/* simple-minded popen2() implementation, returns pid and r/w fds */
static int
popen2(char **cmd, int *pid, int *infd, int *outfd)
{
	int p1[2], p2[2];
	if (!cmd || !pid || !infd || !outfd) return -1;
	if (pipe(p1) == -1) return -1;
	if (pipe(p2) == -1) {
		close(p1[1]); close(p1[0]);
		return -1;
	}
	if ((*pid = fork()) == -1) {
		close(p2[1]); close(p2[0]);
		close(p1[1]); close(p1[0]);
		return -1;
	}
	if (*pid) {
		/* parent */
		*infd = p1[1];
		*outfd = p2[0];
		close(p1[0]); close(p2[1]);
		return 0;
	}

	/* child */
	(void)dup2(p1[0], 0);
	(void)dup2(p2[1], 1);
	close(p1[0]); close(p1[1]);
	close(p2[0]); close(p2[1]);
	execvp(cmd[0], cmd);
	/* oops */
#if DEBUG
perror("execvp");
#endif
	exit(255);
	return -1;
}

/* thread to read from process, write to socket */
long
readthread(void *rts)
{
	int i;
	char x[BUFFERSIZE];
	struct iothreadstate *r = (struct iothreadstate *)rts;

	for(;;) {
		/* drain the entire read buffer */
		while((i = read(r->rfd, &x, BUFFERSIZE)) > 0) {
			(void)send(r->conn_fd, &x, i, 0);
		}

		if (r->pid == -1) break;
		if (has_data(r->rtid)) break; /* signal from main thread */
	}

#if DEBUG
fprintf(stderr, "read thread %d terminating (pid %d)\n", r->rtid, r->pid);
#endif
	return 0;
}

/* thread to read from socket, write to process */
long
writethread(void *wts)
{
	int i;
	char x[BUFFERSIZE];
	struct iothreadstate *w = (struct iothreadstate *)wts;
	fd_set fds, fdss;
	struct timeval tv;

	FD_ZERO(&fds);
	FD_SET(w->conn_fd, &fds);

	for(;;) {
		fdss = fds;
		/* although this is allegedly blocking, the select
			solves certain problems with interrupted system
			calls. however, the select has to be time-limited
			so we know when it's time to check the message
			queue. make this long enough to not overly
			busywait but without excessive latency */
		tv.tv_sec = 0;
		tv.tv_usec = 10000; /* 10ms */
		(void)select(w->conn_fd + 1, &fdss, NULL, NULL, &tv);
		if (FD_ISSET(w->conn_fd, &fdss)) {
			if ((i = recv(w->conn_fd, &x, BUFFERSIZE, 0)) > 0) {
				(void)write(w->wfd, &x, i);
			} else {
				if (errno == EINTR) continue;

				/* disconnect by peer */
#if SDEBUG
/* you should never see this on the socket */
send(w->conn_fd, "bye\n", 4, 0);
#endif
#if DEBUG
fprintf(stderr, "write thread %d disconnected (KILL pid %d) (errno %d)\n",
	w->wtid, w->pid, errno);
#endif
				if (w->pid != -1) send_signal(w->pid, SIGKILL);
				break;
			}
		}

		if (w->pid == -1) break;
		if (has_data(w->wtid)) break;
	}
#if DEBUG
fprintf(stderr, "write thread %d terminating (pid %d)\n", w->wtid, w->pid);
#endif
	return 0;
}

/* cleanup an iothreadstate. assumes process was already killed, or will be.
	this may be called as a separate thread or from the main thread */
long
cleanup(void *cts)
{
	status_t x;
	struct iothreadstate *i = (struct iothreadstate *)cts;

#if SDEBUG
send(i->conn_fd, "terminate!\n", 11, 0);
#endif

	/* killing the threads outright leaves a mess in the net_server, so we
		simply signal them to asynchronously terminate */
	if (i->wtid > 0) { /* write thread first, read may still be xmitting */
		(void)send_data(i->wtid, 1, NULL, 0);
		(void)wait_for_thread(i->wtid, &x);
	}
	(void)close(i->wfd);
	if (i->rtid > 0) {
		(void)send_data(i->rtid, 1, NULL, 0);
		(void)wait_for_thread(i->rtid, &x);
	}
	/* force flush */
	send(i->conn_fd, "", 0, 0);

	(void)close(i->rfd);
	(void)closesocket(i->conn_fd);

	/* release the entry */
	i->pid = -1;
	nsock--;
#if DEBUG
fprintf(stderr, "cleanup thread succeeded r:%d w:%d, %d in flight\n",
	i->rtid, i->wtid, nsock);
#endif
	return 0;
}

int
main(int argc, char **argv)
{
	unsigned short port;
	int listen_fd, conn_fd;
	struct timeval tv;
	struct sockaddr_in sa_in;
	int sz;
	int i;
	char **child_argv;

	if (argc < 3) {
		(void)fprintf(stderr,
			"usage: %s port program [args...]\n", argv[0]);
		exit(255);
	}
	child_argv = argv + 2;

	port = (unsigned short)atoi(argv[1]);
	sz = sizeof(sa_in);
	listen_fd = initialize_listen_socket(PF_INET, AF_INET, port);

	for(i=0; i<NUMSOCKS; i++) { iothreads[i].pid = -1; }
	(void)signal(SIGCHLD, child_handler);
	(void)signal(SIGPIPE, child_handler);

	for(;;) {
		conn_fd = accept(listen_fd, (struct sockaddr*) &sa_in, &sz);
		if (conn_fd < 0) {
			if (errno == EINTR) /* signal */
				continue;
			perror("accept");
			exit(1);
		}
		if (setsockopt(conn_fd, SOL_SOCKET, SO_REUSEADDR,
				(char*) &i, sizeof(i)) < 0) {
			perror("setsockopt");
			(void)closesocket(conn_fd);
			continue;
		}

#if SDEBUG
send(conn_fd, "connect!\n", 9, 0);
#endif

		/* this may happen if we have a free socket in the listen
			backlog but are still closing down a process
			that just died */
		if (nsock == NUMSOCKS) {
			send(conn_fd,
				"max processes, try again later\n", 32, 0);
			(void)closesocket(conn_fd);
			continue;
		}
			
		/* find next available entry */
		nsock++;
		for(csock=0; csock<NUMSOCKS; csock++) {
			if (iothreads[csock].pid == -1) break;
		}
#if DEBUG
fprintf(stderr, "entry %d allocated, %d in flight\n", csock, nsock);
#endif
		iothreads[csock].conn_fd = conn_fd;

		/* launch captive subprocess and start i/o service threads */
		if (!popen2(child_argv,
				&(iothreads[csock].pid),
				&(iothreads[csock].wfd),
				&(iothreads[csock].rfd))) {

			iothreads[csock].rtid = spawn_thread(readthread,
				"inetb read thread",
				B_NORMAL_PRIORITY, &(iothreads[csock]));
			iothreads[csock].wtid = spawn_thread(writethread,
				"inetb write thread",
				B_NORMAL_PRIORITY, &(iothreads[csock]));

			/* out of threads, eek */
			if (iothreads[csock].rtid < 0 ||
					iothreads[csock].wtid < 0) {
				send(conn_fd, "panic: no threads\n", 18, 0);
				fprintf(stderr, "panic: no threads\n");
				(void)send_signal(iothreads[csock].pid,
					SIGKILL);
				/* hope we can clean up gracefully */
				(void)cleanup((void *)&(iothreads[csock]));
				continue;
			}

#if DEBUG
fprintf(stderr, "entry %d: thread start r:%d w:%d servicing pid %d\n",
	csock,
	iothreads[csock].rtid, iothreads[csock].wtid, iothreads[csock].pid);
#endif

			(void)resume_thread(iothreads[csock].rtid);
			(void)resume_thread(iothreads[csock].wtid);
			/* the child reaper handles cleanup on termination */
		} else {
			send(conn_fd, "could not launch process\n", 25, 0);
			(void)closesocket(conn_fd);
			iothreads[csock].pid = -1;
			nsock--;
		}
	}
}

static void
child_handler(int sig)
{
	pid_t pid;
	int i;

	/* keep the signal handlers alive (paranoia) */
	(void)signal(SIGCHLD, child_handler);
	(void)signal(SIGPIPE, child_handler);

#if DEBUG
fprintf(stderr, "child_handler: signal %d\n", sig);
#endif

	/* reap defunct children until there aren't any more */
	for (;;) {
		pid = waitpid((pid_t) -1, &i, WNOHANG);
		if ((int)pid == 0) break; /* done */
		if ((int)pid < 0) {
			if (errno == EINTR) /* hmm */
				continue;
			break;
		}

		/* clear any entries with that pid and return them to pool */
		for (i=0; i<NUMSOCKS; i++) {
			if(iothreads[i].pid == pid) {
				/* launch cleanup on a separate thread */
#if DEBUG
fprintf(stderr, "entry %d: reaped %d, launching cleanup thread\n", i, pid);
#endif

				(void)resume_thread(spawn_thread(cleanup,
					"inetb cleanup thread",
						B_NORMAL_PRIORITY,
							&(iothreads[i])));
			}
		}
	}
}

static int
initialize_listen_socket(int pf, int af, unsigned short port)
{
	int listen_fd;
	int on;
	struct sockaddr_in sa_in;

	listen_fd = socket(af, SOCK_STREAM, 0);
	if (listen_fd < 0) {
		perror("socket");
		exit(1);
	}

	if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR,
		(char*) &on, sizeof(on)) < 0) {
		perror("setsockopt SO_REUSEADDR");
		exit(1);
	}

	(void)memset((char*) &sa_in, 0, sizeof(sa_in));
	sa_in.sin_family = af;
	sa_in.sin_addr.s_addr = htonl(INADDR_ANY);
	sa_in.sin_port = htons(port);

	if (bind(listen_fd, (struct sockaddr*) &sa_in, sizeof(sa_in)) < 0) {
		perror("bind");
		exit(1);
	}

	if (listen(listen_fd, NUMSOCKS) < 0) {
		perror("listen");
		exit(1);
	}

	return listen_fd;
}