pkg/csource: mimic the way syscalls are scheduled in executor

Currently csource uses completely different, simpler way of scheduling
syscalls onto threads (thread per call with random sleeps).
Mimic the way calls are scheduled in executor.

Fixes #312

2 files changed, 70 insertions, 22 deletions

diff --git a/pkg/csource/csource.go b/pkg/csource/csource.go
index 38814c3b2..14f381451 100644
--- a/pkg/csource/csource.go
+++ b/pkg/csource/csource.go
@@ -193,21 +193,17 @@ func (ctx *context) generateTestFunc(calls []string, nvar uint64, name string) {
 		}
 		ctx.printf("}\n\n")
 	} else {
-		ctx.printf("void *thr(void *arg)\n{\n")
-		ctx.printf("\tswitch ((long)arg) {\n")
+		ctx.printf("void execute_call(int call)\n{\n")
+		ctx.printf("\tswitch (call) {\n")
 		for i, c := range calls {
 			ctx.printf("\tcase %v:\n", i)
 			ctx.printf("%s", strings.Replace(c, "\t", "\t\t", -1))
 			ctx.printf("\t\tbreak;\n")
 		}
 		ctx.printf("\t}\n")
-		ctx.printf("\treturn 0;\n}\n\n")
+		ctx.printf("}\n\n")
 
 		ctx.printf("void %v()\n{\n", name)
-		ctx.printf("\tlong i;\n")
-		ctx.printf("\tpthread_t th[%v];\n", 2*len(calls))
-		ctx.printf("\tpthread_attr_t attr;\n")
-		ctx.printf("\n")
 		if opts.Debug {
 			// Use debug to avoid: error: ‘debug’ defined but not used.
 			ctx.printf("\tdebug(\"%v\\n\");\n", name)
@@ -218,23 +214,11 @@ func (ctx *context) generateTestFunc(calls []string, nvar uint64, name string) {
 		if nvar != 0 {
 			ctx.printf("\tmemset(r, -1, sizeof(r));\n")
 		}
+		ctx.printf("\texecute(%v);\n", len(calls))
 		if opts.Collide {
-			ctx.printf("\tsrand(getpid());\n")
-		}
-		ctx.printf("\tpthread_attr_init(&attr);\n")
-		ctx.printf("\tpthread_attr_setstacksize(&attr, 128 << 10);\n")
-		ctx.printf("\tfor (i = 0; i < %v; i++) {\n", len(calls))
-		ctx.printf("\t\tpthread_create(&th[i], &attr, thr, (void*)i);\n")
-		ctx.printf("\t\tusleep(rand()%%10000);\n")
-		ctx.printf("\t}\n")
-		if opts.Collide {
-			ctx.printf("\tfor (i = 0; i < %v; i++) {\n", len(calls))
-			ctx.printf("\t\tpthread_create(&th[%v+i], &attr, thr, (void*)i);\n", len(calls))
-			ctx.printf("\t\tif (rand()%%2)\n")
-			ctx.printf("\t\t\tusleep(rand()%%10000);\n")
-			ctx.printf("\t}\n")
+			ctx.printf("\tcollide = 1;\n")
+			ctx.printf("\texecute(%v);\n", len(calls))
 		}
-		ctx.printf("\tusleep(rand()%%100000);\n")
 		ctx.printf("}\n\n")
 	}
 }
diff --git a/pkg/csource/linux_common.go b/pkg/csource/linux_common.go
index 3ed6fc2f4..1ec165b89 100644
--- a/pkg/csource/linux_common.go
+++ b/pkg/csource/linux_common.go
@@ -12,6 +12,7 @@ var commonHeaderLinux = `
 #include <sys/syscall.h>
 #include <unistd.h>
 #if defined(SYZ_EXECUTOR) || defined(SYZ_THREADED) || defined(SYZ_COLLIDE)
+#include <linux/futex.h>
 #include <pthread.h>
 #include <stdlib.h>
 #endif
@@ -2082,4 +2083,67 @@ void loop()
 }
 #endif
 #endif
+
+#if defined(SYZ_THREADED)
+struct thread_t {
+	int created, running, call;
+	pthread_t th;
+};
+
+static struct thread_t threads[16];
+static void execute_call(int call);
+static int running;
+#if defined(SYZ_COLLIDE)
+static int collide;
+#endif
+
+static void* thr(void* arg)
+{
+	struct thread_t* th = (struct thread_t*)arg;
+	for (;;) {
+		while (!__atomic_load_n(&th->running, __ATOMIC_ACQUIRE))
+			syscall(SYS_futex, &th->running, FUTEX_WAIT, 0, 0);
+		execute_call(th->call);
+		__atomic_fetch_sub(&running, 1, __ATOMIC_RELAXED);
+		__atomic_store_n(&th->running, 0, __ATOMIC_RELEASE);
+		syscall(SYS_futex, &th->running, FUTEX_WAKE);
+	}
+	return 0;
+}
+
+static void execute(int num_calls)
+{
+	int call, thread;
+	running = 0;
+	for (call = 0; call < num_calls; call++) {
+		for (thread = 0; thread < sizeof(threads) / sizeof(threads[0]); thread++) {
+			struct thread_t* th = &threads[thread];
+			if (!th->created) {
+				th->created = 1;
+				pthread_attr_t attr;
+				pthread_attr_init(&attr);
+				pthread_attr_setstacksize(&attr, 128 << 10);
+				pthread_create(&th->th, &attr, thr, th);
+			}
+			if (!__atomic_load_n(&th->running, __ATOMIC_ACQUIRE)) {
+				th->call = call;
+				__atomic_fetch_add(&running, 1, __ATOMIC_RELAXED);
+				__atomic_store_n(&th->running, 1, __ATOMIC_RELEASE);
+				syscall(SYS_futex, &th->running, FUTEX_WAKE);
+#if defined(SYZ_COLLIDE)
+				if (collide && call % 2)
+					break;
+#endif
+				struct timespec ts;
+				ts.tv_sec = 0;
+				ts.tv_nsec = 20 * 1000 * 1000;
+				syscall(SYS_futex, &th->running, FUTEX_WAIT, 1, &ts);
+				if (running)
+					usleep((call == num_calls - 1) ? 10000 : 1000);
+				break;
+			}
+		}
+	}
+}
+#endif
 `