add process main loop to pipe stdout/stderr and logs it

src/Makefile

@@ -24,7 +24,7 @@ libapk.so.$(libapk_soname)-objs := \
 	adb.o adb_comp.o adb_walk_adb.o adb_walk_genadb.o adb_walk_gentext.o adb_walk_text.o apk_adb.o \
 	atom.o balloc.o blob.o commit.o common.o context.o crypto.o crypto_$(CRYPTO).o ctype.o \
 	database.o hash.o extract_v2.o extract_v3.o fs_fsys.o fs_uvol.o \
-	io.o io_gunzip.o io_url_$(URL_BACKEND).o tar.o package.o pathbuilder.o print.o solver.o \
+	io.o io_gunzip.o io_url_$(URL_BACKEND).o tar.o package.o pathbuilder.o print.o process.o solver.o \
 	trust.o version.o
 
 ifneq ($(URL_BACKEND),wget)

src/apk_process.h

+/* apk_process.h - Alpine Package Keeper (APK)
+ *
+ * Copyright (C) 2008-2024 Timo Teräs <timo.teras@iki.fi>
+ * All rights reserved.
+ *
+ * SPDX-License-Identifier: GPL-2.0-only
+ */
+
+#ifndef APK_PROCESS_H
+#define APK_PROCESS_H
+
+#include <sys/types.h>
+#include "apk_defines.h"
+#include "apk_blob.h"
+
+struct apk_out;
+struct apk_istream;
+
+struct apk_process {
+	int pipe_stdin[2], pipe_stdout[2], pipe_stderr[2];
+	pid_t pid;
+	const char *argv0;
+	struct apk_out *out;
+	struct apk_istream *is;
+	apk_blob_t is_blob;
+	unsigned int is_eof : 1;
+	struct buf {
+		uint16_t len;
+		uint8_t buf[1022];
+	} buf_stdout, buf_stderr;
+};
+
+int apk_process_init(struct apk_process *p, const char *argv0, struct apk_out *out, struct apk_istream *is);
+pid_t apk_process_fork(struct apk_process *p);
+int apk_process_run(struct apk_process *p);
+int apk_process_cleanup(struct apk_process *p);
+
+#endif

src/meson.build

@@ -31,6 +31,7 @@ libapk_src = [
 	'package.c',
 	'pathbuilder.c',
 	'print.c',
+	'process.c',
 	'solver.c',
 	'tar.c',
 	'trust.c',

src/process.c

+/* pid.c - Alpine Package Keeper (APK)
+ *
+ * Copyright (C) 2024 Timo Teräs <timo.teras@iki.fi>
+ * All rights reserved.
+ *
+ * SPDX-License-Identifier: GPL-2.0-only
+ */
+
+#include <poll.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/wait.h>
+
+#include "apk_io.h"
+#include "apk_process.h"
+#include "apk_print.h"
+
+static void close_fd(int *fd)
+{
+	if (*fd <= 0) return;
+	close(*fd);
+	*fd = -1;
+}
+
+static void set_non_blocking(int fd)
+{
+	if (fd >= 0) fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
+}
+
+int apk_process_init(struct apk_process *p, const char *argv0, struct apk_out *out, struct apk_istream *is)
+{
+	*p = (struct apk_process) {
+		.argv0 = argv0,
+		.is = is,
+		.out = out,
+	};
+	if (IS_ERR(is)) return -PTR_ERR(is);
+
+	if (is) pipe2(p->pipe_stdin, O_CLOEXEC);
+	else {
+		p->pipe_stdin[0] = open("/dev/null", O_RDONLY);
+		p->pipe_stdin[1] = -1;
+	}
+
+	pipe2(p->pipe_stdout, O_CLOEXEC);
+	pipe2(p->pipe_stderr, O_CLOEXEC);
+
+	set_non_blocking(p->pipe_stdin[1]);
+	set_non_blocking(p->pipe_stdout[0]);
+	set_non_blocking(p->pipe_stderr[0]);
+
+	return 0;
+}
+
+static int buf_process(struct buf *b, int fd, struct apk_out *out, const char *prefix, const char *argv0)
+{
+	ssize_t n = read(fd, &b->buf[b->len], sizeof b->buf - b->len);
+	if (n <= 0) {
+		if (b->len) {
+			apk_out_fmt(out, prefix, "%s: %.*s", argv0, (int)b->len, b->buf);
+			b->len = 0;
+		}
+		return 0;
+	}
+
+	b->len += n;
+
+	uint8_t *pos, *lf, *end = &b->buf[b->len];
+	for (pos = b->buf; (lf = memchr(pos, '\n', end - pos)) != NULL; pos = lf + 1) {
+		apk_out_fmt(out, prefix, "%s: %.*s", argv0, (int)(lf - pos), pos);
+	}
+
+	b->len = end - pos;
+	memmove(b->buf, pos, b->len);
+	return 1;
+}
+
+pid_t apk_process_fork(struct apk_process *p)
+{
+	pid_t pid = fork();
+	if (pid < 0) return pid;
+	if (pid == 0) {
+		dup2(p->pipe_stdin[0], STDIN_FILENO);
+		dup2(p->pipe_stdout[1], STDOUT_FILENO);
+		dup2(p->pipe_stderr[1], STDERR_FILENO);
+		close_fd(&p->pipe_stdin[1]);
+		close_fd(&p->pipe_stdout[0]);
+		close_fd(&p->pipe_stderr[0]);
+		return pid;
+	} else {
+		p->pid = pid;
+	}
+	close_fd(&p->pipe_stdin[0]);
+	close_fd(&p->pipe_stdout[1]);
+	close_fd(&p->pipe_stderr[1]);
+	return pid;
+}
+
+int apk_process_run(struct apk_process *p)
+{
+	struct pollfd fds[3] = {
+		{ .fd = p->pipe_stdout[0], .events = POLLIN },
+		{ .fd = p->pipe_stderr[0], .events = POLLIN },
+		{ .fd = p->pipe_stdin[1],  .events = POLLOUT },
+	};
+
+	while (fds[0].fd >= 0 || fds[1].fd >= 0 || fds[2].fd >= 0) {
+		if (poll(fds, ARRAY_SIZE(fds), -1) <= 0) continue;
+		if (fds[0].revents) {
+			if (!buf_process(&p->buf_stdout, p->pipe_stdout[0], p->out, NULL, p->argv0)) {
+				fds[0].fd = -1;
+				close_fd(&p->pipe_stdout[0]);
+			}
+		}
+		if (fds[1].revents) {
+			if (!buf_process(&p->buf_stderr, p->pipe_stderr[0], p->out, "", p->argv0)) {
+				fds[1].fd = -1;
+				close_fd(&p->pipe_stderr[0]);
+			}
+		}
+		if (fds[2].revents == POLLOUT) {
+			if (!p->is_blob.len) {
+				switch (apk_istream_get_all(p->is, &p->is_blob)) {
+				case 0:
+					break;
+				case -APKE_EOF:
+					p->is_eof = 1;
+					goto stdin_close;
+				default:
+					goto stdin_close;
+				}
+			}
+			int n = write(p->pipe_stdin[1], p->is_blob.ptr, p->is_blob.len);
+			if (n < 0) {
+				if (errno == EWOULDBLOCK) break;
+				goto stdin_close;
+			}
+			p->is_blob.ptr += n;
+			p->is_blob.len -= n;
+		}
+		if (fds[2].revents & POLLERR) {
+		stdin_close:
+			close_fd(&p->pipe_stdin[1]);
+			fds[2].fd = -1;
+		}
+	}
+	return apk_process_cleanup(p);
+}
+
+int apk_process_cleanup(struct apk_process *p)
+{
+	char buf[APK_EXIT_STATUS_MAX_SIZE];
+	int status = 0;
+
+	if (p->is) apk_istream_close(p->is);
+	close_fd(&p->pipe_stdin[1]);
+	close_fd(&p->pipe_stdout[0]);
+	close_fd(&p->pipe_stderr[0]);
+
+	while (waitpid(p->pid, &status, 0) < 0 && errno == EINTR);
+
+	if (apk_exit_status_str(status, buf, sizeof buf)) {
+		apk_err(p->out, "%s: %s", p->argv0, buf);
+		return -1;
+	}
+	if (p->is && !p->is_eof) return -2;
+	return 0;
+}

test/unit/main.c

+#include <signal.h>
 #include "apk_test.h"
 
 static int num_tests;
@@ -13,5 +14,6 @@ void test_register(const char *name, UnitTestFunction f)
 
 int main(void)
 {
+	signal(SIGPIPE, SIG_IGN);
 	return _cmocka_run_group_tests("unit_tests", all_tests, num_tests, NULL, NULL);
 }

test/unit/meson.build

@@ -5,6 +5,7 @@ if cmocka_dep.found()
 unit_test_src = [
 	'blob_test.c',
 	'package_test.c',
+	'process_test.c',
 	'version_test.c',
 	'main.c'
 ]

test/unit/process_test.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include "apk_test.h"
+#include "apk_print.h"
+#include "apk_process.h"
+#include "apk_io.h"
+
+#define writestr(fd, str) write(fd, str, sizeof(str)-1)
+
+struct cached_out {
+	struct apk_out out;
+	char buf_err[256], buf_out[256];
+};
+
+static void open_out(struct cached_out *co)
+{
+	co->out = (struct apk_out) {
+		.out = fmemopen(co->buf_out, sizeof co->buf_out, "w"),
+		.err = fmemopen(co->buf_err, sizeof co->buf_err, "w"),
+	};
+	assert_non_null(co->out.out);
+	assert_non_null(co->out.err);
+}
+
+static void assert_output_equal(struct cached_out *co, const char *expected_err, const char *expected_out)
+{
+	fputc(0, co->out.out);
+	fclose(co->out.out);
+	fputc(0, co->out.err);
+	fclose(co->out.err);
+
+	assert_string_equal(co->buf_err, expected_err);
+	assert_string_equal(co->buf_out, expected_out);
+}
+
+APK_TEST(pid_logging) {
+	struct cached_out co;
+	struct apk_process p;
+
+	open_out(&co);
+	assert_int_equal(0, apk_process_init(&p, "test0", &co.out, NULL));
+	if (apk_process_fork(&p) == 0) {
+		writestr(STDERR_FILENO, "error1\nerror2\n");
+		writestr(STDOUT_FILENO, "hello1\nhello2\n");
+		close(STDOUT_FILENO);
+		usleep(10000);
+		writestr(STDERR_FILENO, "more\nlastline");
+		exit(0);
+	}
+
+	assert_int_equal(0, apk_process_run(&p));
+	assert_output_equal(&co,
+		"test0: error1\n"
+		"test0: error2\n"
+		"test0: more\n"
+		"test0: lastline\n",
+
+		"test0: hello1\n"
+		"test0: hello2\n");
+}
+
+APK_TEST(pid_error_exit) {
+	struct cached_out co;
+	struct apk_process p;
+
+	open_out(&co);
+	assert_int_equal(0, apk_process_init(&p, "test1", &co.out, NULL));
+	if (apk_process_fork(&p) == 0) {
+		exit(100);
+	}
+
+	assert_int_equal(-1, apk_process_run(&p));
+	assert_output_equal(&co,
+		"ERROR: test1: exited with error 100\n",
+		"");
+}
+
+APK_TEST(pid_input_partial) {
+	struct cached_out co;
+	struct apk_process p;
+
+	open_out(&co);
+	assert_int_equal(0, apk_process_init(&p, "test2", &co.out, apk_istream_from_file(AT_FDCWD, "/dev/zero")));
+	if (apk_process_fork(&p) == 0) {
+		char buf[1024];
+		int left = 128*1024;
+		while (left) {
+			int n = read(STDIN_FILENO, buf, min(left, sizeof buf));
+			if (n <= 0) exit(100);
+			left -= n;
+		}
+		writestr(STDOUT_FILENO, "success\n");
+		exit(0);
+	}
+
+	assert_int_equal(-2, apk_process_run(&p));
+	assert_output_equal(&co,
+		"",
+		"test2: success\n");
+}
+
+APK_TEST(pid_input_full) {
+	struct cached_out co;
+	struct apk_process p;
+
+	open_out(&co);
+	assert_int_equal(0, apk_process_init(&p, "test3", &co.out, apk_istream_from_file(AT_FDCWD, "version.data")));
+	if (apk_process_fork(&p) == 0) {
+		char buf[1024];
+		writestr(STDOUT_FILENO, "start reading!\n");
+		usleep(10000);
+		while (1) {
+			int n = read(STDIN_FILENO, buf, sizeof buf);
+			if (n < 0) exit(100);
+			if (n == 0) break;
+		}
+		writestr(STDOUT_FILENO, "success\n");
+		exit(0);
+	}
+
+	assert_int_equal(0, apk_process_run(&p));
+	assert_output_equal(&co,
+		"",
+		"test3: start reading!\n"
+		"test3: success\n");
+}
+
+// FIXME: add test for subprocess _istream