executor, pkg/ipc: unify ipc protocol between linux and other OSes

We currently use more complex and functional protocol on linux,
and a simple ad-hoc protocol on other OSes.
This leads to code duplication in both ipc and executor.
Linux supports coverage, shared memory communication and fork server,
which would also be useful for most other OSes.

Unify communication protocol and parametrize it by
(1) use of shmem or only pipes, (2) use of fork server.

This reduces duplication in ipc and executor and will
allow to support the useful features for other OSes easily.

Finally, this fixes akaros support as it currently uses
syz-stress running on host (linux) and executor running on akaros.

12 files changed, 196 insertions, 189 deletions

diff --git a/executor/common.h b/executor/common.h
index 6cb7635f0..980ea74b7 100644
--- a/executor/common.h
+++ b/executor/common.h
@@ -23,8 +23,15 @@
 #endif
 
 #if defined(SYZ_EXECUTOR)
-#ifndef SYSCALLAPI
+#if defined(__GNUC__)
 #define SYSCALLAPI
+#define NORETURN __attribute__((noreturn))
+#define ALIGNED(N) __attribute__((aligned(N)))
+#else
+// Assuming windows/cl.
+#define SYSCALLAPI WINAPI
+#define NORETURN __declspec(noreturn)
+#define ALIGNED(N) __declspec(align(N))
 #endif
 
 typedef long(SYSCALLAPI* syscall_t)(long, long, long, long, long, long, long, long, long);
@@ -58,7 +65,6 @@ const int kErrorStatus = 68;
 NORETURN static void fail(const char* msg, ...)
 {
 	int e = errno;
-	fflush(stdout);
 	va_list args;
 	va_start(args, msg);
 	vfprintf(stderr, msg, args);
@@ -74,7 +80,6 @@ NORETURN static void fail(const char* msg, ...)
 // kernel error (e.g. wrong syscall return value)
 NORETURN static void error(const char* msg, ...)
 {
-	fflush(stdout);
 	va_list args;
 	va_start(args, msg);
 	vfprintf(stderr, msg, args);
@@ -84,12 +89,11 @@ NORETURN static void error(const char* msg, ...)
 }
 #endif
 
-#if defined(SYZ_EXECUTOR) || (defined(SYZ_REPEAT) && defined(SYZ_WAIT_REPEAT))
+#if defined(SYZ_EXECUTOR) || (defined(SYZ_REPEAT) && defined(SYZ_WAIT_REPEAT) && defined(SYZ_USE_TMP_DIR))
 // just exit (e.g. due to temporal ENOMEM error)
 NORETURN static void exitf(const char* msg, ...)
 {
 	int e = errno;
-	fflush(stdout);
 	va_list args;
 	va_start(args, msg);
 	vfprintf(stderr, msg, args);
diff --git a/executor/common_akaros.h b/executor/common_akaros.h
index 3a12bac0c..d37e05176 100644
--- a/executor/common_akaros.h
+++ b/executor/common_akaros.h
@@ -30,7 +30,6 @@
 #endif
 
 #define doexit exit
-#define NORETURN __attribute__((noreturn))
 
 #include "common.h"
 
diff --git a/executor/common_freebsd.h b/executor/common_freebsd.h
index 87daa17ff..c072b12aa 100644
--- a/executor/common_freebsd.h
+++ b/executor/common_freebsd.h
@@ -19,7 +19,6 @@
 #endif
 
 #define doexit exit
-#define NORETURN __attribute__((noreturn))
 
 #include "common.h"
 
diff --git a/executor/common_fuchsia.h b/executor/common_fuchsia.h
index 0403deafc..4dfad597d 100644
--- a/executor/common_fuchsia.h
+++ b/executor/common_fuchsia.h
@@ -21,7 +21,6 @@
 #endif
 
 #define doexit exit
-#define NORETURN __attribute__((noreturn))
 
 #include "common.h"
 
diff --git a/executor/common_linux.h b/executor/common_linux.h
index df1262107..8b9c00bfd 100644
--- a/executor/common_linux.h
+++ b/executor/common_linux.h
@@ -127,7 +127,6 @@ __attribute__((noreturn)) static void doexit(int status)
 	for (i = 0;; i++) {
 	}
 }
-#define NORETURN __attribute__((noreturn))
 #endif
 
 #if defined(SYZ_EXECUTOR)
diff --git a/executor/common_windows.h b/executor/common_windows.h
index dbc6b67c3..077bc874a 100644
--- a/executor/common_windows.h
+++ b/executor/common_windows.h
@@ -6,8 +6,6 @@
 #include <windows.h>
 
 #define doexit exit
-#define NORETURN
-#define SYSCALLAPI WINAPI
 
 #include "common.h"
 
diff --git a/executor/executor.h b/executor/executor.h
index 94d3a2ca0..9e56f45e2 100644
--- a/executor/executor.h
+++ b/executor/executor.h
@@ -16,6 +16,19 @@
 #define GIT_REVISION "unknown"
 #endif
 
+#ifndef GOOS
+#define GOOS "unknown"
+#endif
+
+// Note: zircon max fd is 256.
+#ifndef DUP2_BROKEN
+const int kInPipeFd = 250; // remapped from stdin
+const int kOutPipeFd = 251; // remapped from stdout
+#else
+const int kInPipeFd = 0;
+const int kOutPipeFd = 1;
+#endif
+
 const int kMaxInput = 2 << 20;
 const int kMaxOutput = 16 << 20;
 const int kMaxArgs = 9;
@@ -56,10 +69,15 @@ bool flag_inject_fault;
 int flag_fault_call;
 int flag_fault_nth;
 
+int flag_pid;
+
 int running;
 uint32_t completed;
 bool collide;
 
+ALIGNED(64 << 10)
+char input_data[kMaxInput];
+
 // We use the default value instead of results of failed syscalls.
 // -1 is an invalid fd and an invalid address and deterministic,
 // so good enough for our purposes.
@@ -106,6 +124,35 @@ struct res_t {
 
 res_t results[kMaxCommands];
 
+const uint64_t kInMagic = 0xbadc0ffeebadface;
+const uint32_t kOutMagic = 0xbadf00d;
+
+struct handshake_req {
+	uint64_t magic;
+	uint64_t flags; // env flags
+	uint64_t pid;
+};
+
+struct handshake_reply {
+	uint32_t magic;
+};
+
+struct execute_req {
+	uint64_t magic;
+	uint64_t env_flags;
+	uint64_t exec_flags;
+	uint64_t pid;
+	uint64_t fault_call;
+	uint64_t fault_nth;
+	uint64_t prog_size;
+};
+
+struct execute_reply {
+	uint32_t magic;
+	uint32_t done;
+	uint32_t status;
+};
+
 enum {
 	KCOV_CMP_CONST = 1,
 	KCOV_CMP_SIZE1 = 0,
@@ -148,10 +195,107 @@ uint64_t read_cover_size(thread_t* th);
 static uint32_t hash(uint32_t a);
 static bool dedup(uint32_t sig);
 
-void execute_one(uint64_t* input_data)
+void setup_control_pipes()
+{
+#ifndef DUP2_BROKEN
+	if (dup2(0, kInPipeFd) < 0)
+		fail("dup2(0, kInPipeFd) failed");
+	if (dup2(1, kOutPipeFd) < 0)
+		fail("dup2(1, kOutPipeFd) failed");
+	if (dup2(2, 1) < 0)
+		fail("dup2(2, 1) failed");
+	if (close(0))
+		fail("close(0) failed");
+#endif
+}
+
+void parse_env_flags(uint64_t flags)
+{
+	flag_debug = flags & (1 << 0);
+	flag_cover = flags & (1 << 1);
+	flag_threaded = flags & (1 << 2);
+	flag_collide = flags & (1 << 3);
+	flag_sandbox = sandbox_none;
+	if (flags & (1 << 4))
+		flag_sandbox = sandbox_setuid;
+	else if (flags & (1 << 5))
+		flag_sandbox = sandbox_namespace;
+	if (!flag_threaded)
+		flag_collide = false;
+	flag_enable_tun = flags & (1 << 6);
+	flag_enable_fault_injection = flags & (1 << 7);
+}
+
+void receive_handshake()
+{
+	handshake_req req = {};
+	int n = read(kInPipeFd, &req, sizeof(req));
+	if (n != sizeof(req))
+		fail("handshake read failed: %d", n);
+	if (req.magic != kInMagic)
+		fail("bad handshake magic 0x%llx", req.magic);
+	parse_env_flags(req.flags);
+	flag_pid = req.pid;
+}
+
+void reply_handshake()
+{
+	handshake_reply reply = {};
+	reply.magic = kOutMagic;
+	if (write(kOutPipeFd, &reply, sizeof(reply)) != sizeof(reply))
+		fail("control pipe write failed");
+}
+
+void receive_execute()
+{
+	execute_req req;
+	if (read(kInPipeFd, &req, sizeof(req)) != (ssize_t)sizeof(req))
+		fail("control pipe read failed");
+	if (req.magic != kInMagic)
+		fail("bad execute request magic 0x%llx", req.magic);
+	if (req.prog_size > kMaxInput)
+		fail("bad execute prog size 0x%llx", req.prog_size);
+	parse_env_flags(req.env_flags);
+	flag_pid = req.pid;
+	flag_collect_cover = req.exec_flags & (1 << 0);
+	flag_dedup_cover = req.exec_flags & (1 << 1);
+	flag_inject_fault = req.exec_flags & (1 << 2);
+	flag_collect_comps = req.exec_flags & (1 << 3);
+	flag_fault_call = req.fault_call;
+	flag_fault_nth = req.fault_nth;
+	debug("exec opts: cover=%d comps=%d dedup=%d fault=%d/%d/%d\n",
+	      flag_collect_cover, flag_collect_comps, flag_dedup_cover,
+	      flag_inject_fault, flag_fault_call, flag_fault_nth);
+	if (req.prog_size == 0)
+		return;
+	uint64_t pos = 0;
+	for (;;) {
+		ssize_t rv = read(kInPipeFd, input_data + pos, sizeof(input_data) - pos);
+		if (rv < 0)
+			fail("read failed");
+		pos += rv;
+		if (rv == 0 || pos >= req.prog_size)
+			break;
+	}
+	if (pos != req.prog_size)
+		fail("bad input size %d, want %d", pos, req.prog_size);
+}
+
+void reply_execute(int status)
+{
+	execute_reply reply = {};
+	reply.magic = kOutMagic;
+	reply.done = true;
+	reply.status = status;
+	if (write(kOutPipeFd, &reply, sizeof(reply)) != sizeof(reply))
+		fail("control pipe write failed");
+}
+
+// execute_one executes program stored in input_data.
+void execute_one()
 {
 retry:
-	uint64_t* input_pos = input_data;
+	uint64_t* input_pos = (uint64_t*)input_data;
 	write_output(0); // Number of executed syscalls (updated later).
 
 	if (!collide && !flag_threaded)
@@ -611,8 +755,8 @@ uint64_t read_result(uint64_t** input_posp)
 uint64_t read_input(uint64_t** input_posp, bool peek)
 {
 	uint64_t* input_pos = *input_posp;
-	//if ((char*)input_pos >= input_data + kMaxInput)
-	//	fail("input command overflows input");
+	if ((char*)input_pos >= input_data + kMaxInput)
+		fail("input command overflows input");
 	if (!peek)
 		*input_posp = input_pos + 1;
 	return *input_pos;
diff --git a/executor/executor_akaros.cc b/executor/executor_akaros.cc
index 15eafba71..e7d828a40 100644
--- a/executor/executor_akaros.cc
+++ b/executor/executor_akaros.cc
@@ -3,6 +3,9 @@
 
 // +build
 
+// https://github.com/brho/akaros/issues/41
+#define DUP2_BROKEN
+
 #define SYZ_EXECUTOR
 #include "common_akaros.h"
 
@@ -12,83 +15,42 @@
 
 #include "syscalls_akaros.h"
 
-char input_buffer[kMaxInput];
 uint32_t output;
 
-struct in_header {
-	uint64_t magic;
-	uint64_t flags;
-	uint64_t pid;
-	uint64_t progSize;
-	uint64_t execFlags;
-	uint64_t prog[0];
-};
-
-struct out_header {
-	uint64_t magic;
-	uint64_t status;
-};
-
-const uint64_t kInMagic = 0xbadc0ffee;
-const uint64_t kOutMagic = 0xbadf00d;
-
 int main(int argc, char** argv)
 {
 	if (argc == 2 && strcmp(argv[1], "version") == 0) {
-		puts("akaros " GOARCH " " SYZ_REVISION " " GIT_REVISION);
+		puts(GOOS " " GOARCH " " SYZ_REVISION " " GIT_REVISION);
 		return 0;
 	}
 
 	use_temporary_dir();
 	install_segv_handler();
+	setup_control_pipes();
+	receive_handshake();
+	reply_handshake();
+
 	for (;;) {
-		size_t pos = 0;
-		in_header* hdr = (in_header*)input_buffer;
-		for (;;) {
-			int rv = read(0, input_buffer + pos, sizeof(input_buffer) - pos);
-			if (rv < 0)
-				fail("read failed");
-			if (rv == 0)
-				fail("stdin closed, read %d", (int)pos);
-			pos += rv;
-			if (pos > sizeof(in_header)) {
-				if (hdr->magic != kInMagic)
-					fail("bad header magic 0x%llx", hdr->magic);
-				if (pos > sizeof(in_header) + hdr->progSize)
-					fail("excessive input data");
-				if (pos == sizeof(in_header) + hdr->progSize)
-					break;
-			}
-		}
-		flag_debug = hdr->flags & (1 << 0);
-		flag_threaded = hdr->flags & (1 << 2);
-		flag_collide = hdr->flags & (1 << 3);
-		if (!flag_threaded)
-			flag_collide = false;
-		debug("input %d, threaded=%d collide=%d pid=%llu\n",
-		      pos, flag_threaded, flag_collide, hdr->pid);
+		receive_execute();
 		char cwdbuf[128] = "/syz-tmpXXXXXX";
 		mkdtemp(cwdbuf);
 		int pid = fork();
 		if (pid < 0)
 			fail("fork failed");
 		if (pid == 0) {
-			close(0);
-			dup2(2, 1);
+			close(kInPipeFd);
+			close(kOutPipeFd);
 			if (chdir(cwdbuf))
 				fail("chdir failed");
-			execute_one(hdr->prog);
+			execute_one();
 			doexit(0);
 		}
+		// TODO: timeout.
 		int status = 0;
 		while (waitpid(pid, &status, 0) != pid) {
 		}
 		remove_dir(cwdbuf);
-		out_header out;
-		out.magic = kOutMagic;
-		out.status = 0;
-		if (write(1, &out, sizeof(out)) != sizeof(out))
-			fail("stdout write failed");
+		reply_execute(0);
 	}
 	return 0;
 }
diff --git a/executor/executor_freebsd.cc b/executor/executor_freebsd.cc
index 8435d4465..abfb4139b 100644
--- a/executor/executor_freebsd.cc
+++ b/executor/executor_freebsd.cc
@@ -16,13 +16,12 @@
 #include <sys/time.h>
 #include <sys/types.h>
 
-char input_data[kMaxInput];
 uint32_t output;
 
 int main(int argc, char** argv)
 {
 	if (argc == 2 && strcmp(argv[1], "version") == 0) {
-		puts("linux " GOARCH " " SYZ_REVISION " " GIT_REVISION);
+		puts(GOOS " " GOARCH " " SYZ_REVISION " " GIT_REVISION);
 		return 0;
 	}
 
@@ -40,29 +39,9 @@ int main(int argc, char** argv)
 	setrlimit(RLIMIT_CORE, &rlim);
 
 	install_segv_handler();
-	int pos = 0;
-	for (;;) {
-		int rv = read(0, input_data + pos, sizeof(input_data) - pos);
-		if (rv < 0)
-			fail("read failed");
-		if (rv == 0)
-			break;
-		pos += rv;
-	}
-	if (pos < 24)
-		fail("truncated input");
-
-	uint64_t flags = *(uint64_t*)input_data;
-	flag_debug = flags & (1 << 0);
-	flag_threaded = flags & (1 << 2);
-	flag_collide = flags & (1 << 3);
-	if (!flag_threaded)
-		flag_collide = false;
-	uint64_t executor_pid = *((uint64_t*)input_data + 2);
-	debug("input %d, threaded=%d collide=%d pid=%llu\n",
-	      pos, flag_threaded, flag_collide, executor_pid);
-
-	execute_one(((uint64_t*)input_data) + 3);
+	setup_control_pipes();
+	receive_execute();
+	execute_one();
 	return 0;
 }
 
diff --git a/executor/executor_fuchsia.cc b/executor/executor_fuchsia.cc
index a5aa19470..dd9204d7d 100644
--- a/executor/executor_fuchsia.cc
+++ b/executor/executor_fuchsia.cc
@@ -12,40 +12,19 @@
 
 #include "syscalls_fuchsia.h"
 
-char input_data[kMaxInput];
 uint32_t output;
 
 int main(int argc, char** argv)
 {
 	if (argc == 2 && strcmp(argv[1], "version") == 0) {
-		puts("linux " GOARCH " " SYZ_REVISION " " GIT_REVISION);
+		puts(GOOS " " GOARCH " " SYZ_REVISION " " GIT_REVISION);
 		return 0;
 	}
 
 	install_segv_handler();
-	int pos = 0;
-	for (;;) {
-		int rv = read(0, input_data + pos, sizeof(input_data) - pos);
-		if (rv < 0)
-			fail("read failed");
-		if (rv == 0)
-			break;
-		pos += rv;
-	}
-	if (pos < 24)
-		fail("truncated input");
-
-	uint64_t flags = *(uint64_t*)input_data;
-	flag_debug = flags & (1 << 0);
-	flag_threaded = flags & (1 << 2);
-	flag_collide = flags & (1 << 3);
-	if (!flag_threaded)
-		flag_collide = false;
-	uint64_t executor_pid = *((uint64_t*)input_data + 2);
-	debug("input %d, threaded=%d collide=%d pid=%llu\n",
-	      pos, flag_threaded, flag_collide, executor_pid);
-
-	execute_one(((uint64_t*)input_data) + 3);
+	setup_control_pipes();
+	receive_execute();
+	execute_one();
 	return 0;
 }
 
diff --git a/executor/executor_linux.cc b/executor/executor_linux.cc
index 7a26d71c0..72bd91aa8 100644
--- a/executor/executor_linux.cc
+++ b/executor/executor_linux.cc
@@ -34,19 +34,16 @@ const unsigned long KCOV_TRACE_CMP = 1;
 
 const int kInFd = 3;
 const int kOutFd = 4;
-const int kInPipeFd = 5;
-const int kOutPipeFd = 6;
 const int kCoverSize = 64 << 10;
 const int kPageSize = 4 << 10;
 
-__attribute__((aligned(64 << 10))) char input_data[kMaxInput];
 uint32_t* output_data;
 uint32_t* output_pos;
 
 int main(int argc, char** argv)
 {
 	if (argc == 2 && strcmp(argv[1], "version") == 0) {
-		puts("linux " GOARCH " " SYZ_REVISION " " GIT_REVISION);
+		puts(GOOS " " GOARCH " " SYZ_REVISION " " GIT_REVISION);
 		return 0;
 	}
 
@@ -67,23 +64,9 @@ int main(int argc, char** argv)
 	// That's also the reason why we close kInPipeFd/kOutPipeFd below.
 	close(kInFd);
 	close(kOutFd);
+	setup_control_pipes();
+	receive_handshake();
 
-	uint64_t flags = *(uint64_t*)input_data;
-	flag_debug = flags & (1 << 0);
-	flag_cover = flags & (1 << 1);
-	flag_threaded = flags & (1 << 2);
-	flag_collide = flags & (1 << 3);
-	flag_sandbox = sandbox_none;
-	if (flags & (1 << 4))
-		flag_sandbox = sandbox_setuid;
-	else if (flags & (1 << 5))
-		flag_sandbox = sandbox_namespace;
-	if (!flag_threaded)
-		flag_collide = false;
-	flag_enable_tun = flags & (1 << 6);
-	flag_enable_fault_injection = flags & (1 << 7);
-
-	uint64_t executor_pid = *((uint64_t*)input_data + 1);
 	cover_open();
 	install_segv_handler();
 	use_temporary_dir();
@@ -101,13 +84,13 @@ int main(int argc, char** argv)
 	int pid = -1;
 	switch (flag_sandbox) {
 	case sandbox_none:
-		pid = do_sandbox_none(executor_pid, flag_enable_tun);
+		pid = do_sandbox_none(flag_pid, flag_enable_tun);
 		break;
 	case sandbox_setuid:
-		pid = do_sandbox_setuid(executor_pid, flag_enable_tun);
+		pid = do_sandbox_setuid(flag_pid, flag_enable_tun);
 		break;
 	case sandbox_namespace:
-		pid = do_sandbox_namespace(executor_pid, flag_enable_tun);
+		pid = do_sandbox_namespace(flag_pid, flag_enable_tun);
 		break;
 	default:
 		fail("unknown sandbox type");
@@ -119,15 +102,14 @@ int main(int argc, char** argv)
 	while (waitpid(-1, &status, __WALL) != pid) {
 	}
 	status = WEXITSTATUS(status);
+	if (status == 0)
+		status = kRetryStatus;
 	// If an external sandbox process wraps executor, the out pipe will be closed
 	// before the sandbox process exits this will make ipc package kill the sandbox.
 	// As the result sandbox process will exit with exit status 9 instead of the executor
 	// exit status (notably kRetryStatus). Consequently, ipc will treat it as hard
 	// failure rather than a temporal failure. So we duplicate the exit status on the pipe.
-	char tmp = status;
-	if (write(kOutPipeFd, &tmp, 1)) {
-		// Not much we can do, but gcc wants us to check the return value.
-	}
+	reply_execute(status);
 	errno = 0;
 	if (status == kFailStatus)
 		fail("loop failed");
@@ -144,9 +126,7 @@ int main(int argc, char** argv)
 void loop()
 {
 	// Tell parent that we are ready to serve.
-	char tmp = 0;
-	if (write(kOutPipeFd, &tmp, 1) != 1)
-		fail("control pipe write failed");
+	reply_handshake();
 
 	for (int iter = 0;; iter++) {
 		// Create a new private work dir for this test (removed at the end of the loop).
@@ -158,19 +138,7 @@ void loop()
 		// TODO: consider moving the read into the child.
 		// Potentially it can speed up things a bit -- when the read finishes
 		// we already have a forked worker process.
-		uint64_t in_cmd[3] = {};
-		if (read(kInPipeFd, &in_cmd[0], sizeof(in_cmd)) != (ssize_t)sizeof(in_cmd))
-			fail("control pipe read failed");
-		flag_collect_cover = in_cmd[0] & (1 << 0);
-		flag_dedup_cover = in_cmd[0] & (1 << 1);
-		flag_inject_fault = in_cmd[0] & (1 << 2);
-		flag_collect_comps = in_cmd[0] & (1 << 3);
-		flag_fault_call = in_cmd[1];
-		flag_fault_nth = in_cmd[2];
-		debug("exec opts: cover=%d comps=%d dedup=%d fault=%d/%d/%d\n", flag_collect_cover,
-		      flag_collect_comps, flag_dedup_cover,
-		      flag_inject_fault, flag_fault_call, flag_fault_nth);
-
+		receive_execute();
 		int pid = fork();
 		if (pid < 0)
 			fail("clone failed");
@@ -186,9 +154,8 @@ void loop()
 				// isolate consequently executing programs.
 				flush_tun();
 			}
-			uint64_t* input_pos = ((uint64_t*)&input_data[0]) + 2; // skip flags and pid
 			output_pos = output_data;
-			execute_one(input_pos);
+			execute_one();
 			debug("worker exiting\n");
 			doexit(0);
 		}
@@ -247,8 +214,7 @@ void loop()
 		if (status == kErrorStatus)
 			error("child errored");
 		remove_dir(cwdbuf);
-		if (write(kOutPipeFd, &tmp, 1) != 1)
-			fail("control pipe write failed");
+		reply_execute(0);
 	}
 }
 
diff --git a/executor/executor_windows.cc b/executor/executor_windows.cc
index 862621951..0124aa0d9 100644
--- a/executor/executor_windows.cc
+++ b/executor/executor_windows.cc
@@ -14,39 +14,18 @@
 
 #include "syscalls_windows.h"
 
-char input_data[kMaxInput];
 uint32_t output;
 
 int main(int argc, char** argv)
 {
 	if (argc == 2 && strcmp(argv[1], "version") == 0) {
-		puts("linux " GOARCH " " SYZ_REVISION " " GIT_REVISION);
+		puts(GOOS " " GOARCH " " SYZ_REVISION " " GIT_REVISION);
 		return 0;
 	}
 
-	int pos = 0;
-	for (;;) {
-		int rv = _read(0, input_data + pos, sizeof(input_data) - pos);
-		if (rv < 0)
-			fail("read failed");
-		if (rv == 0)
-			break;
-		pos += rv;
-	}
-	if (pos < 24)
-		fail("truncated input");
-
-	uint64_t flags = *(uint64_t*)input_data;
-	flag_debug = flags & (1 << 0);
-	flag_threaded = flags & (1 << 2);
-	flag_collide = flags & (1 << 3);
-	if (!flag_threaded)
-		flag_collide = false;
-	uint64_t executor_pid = *((uint64_t*)input_data + 2);
-	debug("input %d, threaded=%d collide=%d pid=%llu\n",
-	      pos, flag_threaded, flag_collide, executor_pid);
-
-	execute_one(((uint64_t*)input_data) + 3);
+	setup_control_pipes();
+	receive_execute();
+	execute_one();
 	return 0;
 }