Merge pull request #203 from gwsystems/smart-sigalrm

Smart sigalrm

.vscode/settings.json

@@ -53,7 +53,10 @@
 		"string.h": "c",
 		"errno.h": "c",
 		"siginfo_t.h": "c",
-		"features.h": "c"
+		"features.h": "c",
+		"time.h": "c",
+		"local_runqueue_minheap.h": "c",
+		"global_request_scheduler.h": "c"
 	},
 	"files.exclude": {
 		"**/.git": true,

runtime/experiments/bash_libraries/framework.sh

@@ -297,6 +297,16 @@ __framework_sh__load_env_file() {
 	fi
 }
 
+__framework_sh__unset_env_file() {
+	local envfile="$1"
+	if [[ -f "$envfile" ]]; then
+		while read -r line; do
+			echo unset "${line//=*/}"
+			unset "${line//=*/}"
+		done < "$envfile"
+	fi
+}
+
 __framework_sh__run_both() {
 	local short_name
 	for envfile in "$__framework_sh__application_directory"/*.env; do
@@ -321,6 +331,8 @@ __framework_sh__run_both() {
 			return 1
 		}
 
+		__framework_sh__unset_env_file "$envfile"
+
 	done
 
 	return 0

runtime/experiments/bimodal/install.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	HEY_URL=https://hey-release.s3.us-east-2.amazonaws.com/hey_linux_amd64
 	wget $HEY_URL -O hey
 	chmod +x hey

runtime/experiments/bimodal/run.sh

@@ -19,7 +19,7 @@ source get_result_count.sh || exit 1
 source panic.sh || exit 1
 source path_join.sh || exit 1
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	echo "hey is not present."
 	exit 1
 fi

runtime/experiments/concurrency/install.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	HEY_URL=https://hey-release.s3.us-east-2.amazonaws.com/hey_linux_amd64
 	wget $HEY_URL -O hey
 	chmod +x hey

runtime/experiments/concurrency/run.sh

@@ -15,7 +15,7 @@ source get_result_count.sh || exit 1
 source panic.sh || exit 1
 source path_join.sh || exit 1
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	echo "hey is not present."
 	exit 1
 fi

runtime/experiments/payload/install.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	HEY_URL=https://hey-release.s3.us-east-2.amazonaws.com/hey_linux_amd64
 	wget $HEY_URL -O hey
 	chmod +x hey

runtime/experiments/payload/run.sh

@@ -16,7 +16,7 @@ source "framework.sh" || exit 1
 source "get_result_count.sh" || exit 1
 source "generate_gnuplots.sh" || exit 1
 
-if ! command -v hey; then
+if ! command -v hey > /dev/null; then
 	echo "hey is not present."
 	exit 1
 fi

runtime/include/runtime.h

@@ -28,12 +28,30 @@
 /* 100 MB */
 #define RUNTIME_HTTP_RESPONSE_SIZE_MAX 100000000
 
+/* Static buffer used for global deadline array */
+#define RUNTIME_MAX_WORKER_COUNT 32
+
+#ifndef NCORES
+#warning "NCORES not defined in Makefile. Defaulting to 2"
+#define NCORES 2
+#endif
+
+#if NCORES == 1
+#error "RUNTIME MINIMUM REQUIREMENT IS 2 CORES"
+#endif
+
+#define RUNTIME_WORKER_THREAD_CORE_COUNT (NCORES > 1 ? NCORES - 1 : NCORES)
+
 /*
  * Descriptor of the epoll instance used to monitor the socket descriptors of registered
  * serverless modules. The listener cores listens for incoming client requests through this.
  */
 extern int runtime_epoll_file_descriptor;
 
+extern int runtime_worker_threads_argument[RUNTIME_WORKER_THREAD_CORE_COUNT];
+
+extern uint64_t runtime_worker_threads_deadline[RUNTIME_WORKER_THREAD_CORE_COUNT];
+
 /* Optional path to a file to log sandbox perf metrics */
 extern FILE *runtime_sandbox_perf_log;
 
@@ -83,12 +101,31 @@ print_runtime_scheduler(enum RUNTIME_SCHEDULER variant)
 {
 	switch (variant) {
 	case RUNTIME_SCHEDULER_FIFO:
-		return "RUNTIME_SCHEDULER_FIFO";
+		return "FIFO";
 	case RUNTIME_SCHEDULER_EDF:
-		return "RUNTIME_SCHEDULER_EDF";
+		return "EDF";
+	}
+};
+
+enum RUNTIME_SIGALRM_HANDLER
+{
+	RUNTIME_SIGALRM_HANDLER_BROADCAST = 0,
+	RUNTIME_SIGALRM_HANDLER_TRIAGED   = 1
+};
+
+
+static inline char *
+print_runtime_sigalrm_handler(enum RUNTIME_SIGALRM_HANDLER variant)
+{
+	switch (variant) {
+	case RUNTIME_SIGALRM_HANDLER_BROADCAST:
+		return "BROADCAST";
+	case RUNTIME_SIGALRM_HANDLER_TRIAGED:
+		return "TRIAGED";
 	}
 };
 
 extern enum RUNTIME_SCHEDULER       runtime_scheduler;
+extern enum RUNTIME_SIGALRM_HANDLER runtime_sigalrm_handler;
 extern bool                         runtime_preemption_enabled;
 extern uint32_t                     runtime_quantum_us;

runtime/include/worker_thread.h

@@ -2,20 +2,10 @@
 
 #include "runtime.h"
 
-#ifndef NCORES
-#warning "NCORES not defined in Makefile. Defaulting to 2"
-#define NCORES 2
-#endif
-
-#if NCORES == 1
-#error "RUNTIME MINIMUM REQUIREMENT IS 2 CORES"
-#endif
-
-#define WORKER_THREAD_CORE_COUNT (NCORES > 1 ? NCORES - 1 : NCORES)
-
 extern __thread uint64_t worker_thread_lock_duration;
 extern __thread uint64_t worker_thread_start_timestamp;
 extern __thread int      worker_thread_epoll_file_descriptor;
+extern __thread int      worker_thread_idx;
 
 void *worker_thread_main(void *return_code);
 

runtime/src/local_runqueue_minheap.c

@@ -9,6 +9,7 @@
 #include "panic.h"
 #include "priority_queue.h"
 #include "software_interrupt.h"
+#include "runtime.h"
 
 __thread static struct priority_queue *local_runqueue_minheap;
 
@@ -185,6 +186,7 @@ local_runqueue_minheap_preempt(ucontext_t *user_context)
 		 * user-level context switch state, so do not enable software interrupts.
 		 * TODO: Review the interrupt logic here. Issue #63
 		 */
+		runtime_worker_threads_deadline[worker_thread_idx] = next_sandbox->absolute_deadline;
 		arch_context_restore_new(&user_context->uc_mcontext, &next_sandbox->ctxt);
 		should_enable_software_interrupt = false;
 	}

runtime/src/main.c

@@ -30,14 +30,12 @@ uint32_t       runtime_processor_speed_MHz     = 0;
 uint32_t       runtime_total_online_processors = 0;
 uint32_t       runtime_worker_threads_count    = 0;
 const uint32_t runtime_first_worker_processor  = 1;
-/* TODO: the worker never actually records state here */
-int runtime_worker_threads_argument[WORKER_THREAD_CORE_COUNT] = { 0 }; /* The worker sets its argument to -1 on error */
-pthread_t runtime_worker_threads[WORKER_THREAD_CORE_COUNT];
 
 
 FILE *runtime_sandbox_perf_log = NULL;
 
-enum RUNTIME_SCHEDULER runtime_scheduler = RUNTIME_SCHEDULER_FIFO;
+enum RUNTIME_SCHEDULER       runtime_scheduler       = RUNTIME_SCHEDULER_EDF;
+enum RUNTIME_SIGALRM_HANDLER runtime_sigalrm_handler = RUNTIME_SIGALRM_HANDLER_BROADCAST;
 int                          runtime_worker_core_count;
 
 
@@ -175,6 +173,8 @@ runtime_start_runtime_worker_threads()
 {
 	printf("Starting %d worker thread(s)\n", runtime_worker_threads_count);
 	for (int i = 0; i < runtime_worker_threads_count; i++) {
+		/* Pass the value we want the threads to use when indexing into global arrays of per-thread values */
+		runtime_worker_threads_argument[i] = i;
 		int ret = pthread_create(&runtime_worker_threads[i], NULL, worker_thread_main,
 		                         (void *)&runtime_worker_threads_argument[i]);
 		if (ret) {
@@ -192,22 +192,12 @@ runtime_start_runtime_worker_threads()
 	debuglog("Sandboxing environment ready!\n");
 }
 
-void
-runtime_cleanup()
-{
-	if (runtime_sandbox_perf_log != NULL) fflush(runtime_sandbox_perf_log);
-
-	exit(EXIT_SUCCESS);
-}
-
 void
 runtime_configure()
 {
-	signal(SIGTERM, runtime_cleanup);
 	/* Scheduler Policy */
 	char *scheduler_policy = getenv("SLEDGE_SCHEDULER");
-	if (scheduler_policy == NULL) scheduler_policy = "FIFO";
+	if (scheduler_policy == NULL) scheduler_policy = "EDF";
-
 	if (strcmp(scheduler_policy, "EDF") == 0) {
 		runtime_scheduler = RUNTIME_SCHEDULER_EDF;
 	} else if (strcmp(scheduler_policy, "FIFO") == 0) {
@@ -217,6 +207,20 @@ runtime_configure()
 	}
 	printf("\tScheduler Policy: %s\n", print_runtime_scheduler(runtime_scheduler));
 
+	/* Sigalrm Handler Technique */
+	char *sigalrm_policy = getenv("SLEDGE_SIGALRM_HANDLER");
+	if (sigalrm_policy == NULL) sigalrm_policy = "BROADCAST";
+	if (strcmp(sigalrm_policy, "BROADCAST") == 0) {
+		runtime_sigalrm_handler = RUNTIME_SIGALRM_HANDLER_BROADCAST;
+	} else if (strcmp(sigalrm_policy, "TRIAGED") == 0) {
+		if (unlikely(runtime_scheduler != RUNTIME_SCHEDULER_EDF))
+			panic("triaged sigalrm handlers are only valid with EDF\n");
+		runtime_sigalrm_handler = RUNTIME_SIGALRM_HANDLER_TRIAGED;
+	} else {
+		panic("Invalid sigalrm policy: %s. Must be {BROADCAST|TRIAGED}\n", sigalrm_policy);
+	}
+	printf("\tSigalrm Policy: %s\n", print_runtime_sigalrm_handler(runtime_sigalrm_handler));
+
 	/* Runtime Preemption Toggle */
 	char *preempt_disable = getenv("SLEDGE_DISABLE_PREEMPTION");
 	if (preempt_disable != NULL && strcmp(preempt_disable, "false") != 0) runtime_preemption_enabled = false;
@@ -367,7 +371,7 @@ main(int argc, char **argv)
 
 	printf("Runtime Environment:\n");
 
-	memset(runtime_worker_threads, 0, sizeof(pthread_t) * WORKER_THREAD_CORE_COUNT);
+	memset(runtime_worker_threads, 0, sizeof(pthread_t) * RUNTIME_WORKER_THREAD_CORE_COUNT);
 
 	runtime_processor_speed_MHz = runtime_get_processor_speed_MHz();
 	if (unlikely(runtime_processor_speed_MHz == 0)) panic("Failed to detect processor speed\n");

runtime/src/runtime.c

@@ -21,11 +21,23 @@
  **************************/
 
 int       runtime_epoll_file_descriptor;
+pthread_t runtime_worker_threads[RUNTIME_WORKER_THREAD_CORE_COUNT];
+int       runtime_worker_threads_argument[RUNTIME_WORKER_THREAD_CORE_COUNT] = { 0 };
+/* The active deadline of the sandbox running on each worker thread */
+uint64_t runtime_worker_threads_deadline[RUNTIME_WORKER_THREAD_CORE_COUNT] = { UINT64_MAX };
 
 /******************************************
  * Shared Process / Listener Thread Logic *
  *****************************************/
 
+void
+runtime_cleanup()
+{
+	if (runtime_sandbox_perf_log != NULL) fflush(runtime_sandbox_perf_log);
+
+	exit(EXIT_SUCCESS);
+}
+
 /**
  * Initialize runtime global state, mask signals, and init http parser
  */
@@ -52,10 +64,12 @@ runtime_initialize(void)
 		panic("Invalid scheduler policy set: %u\n", runtime_scheduler);
 	}
 
-	/* Mask Signals */
+	/* Configure Signals */
+	signal(SIGPIPE, SIG_IGN);
+	signal(SIGTERM, runtime_cleanup);
+	/* These should only be unmasked by workers */
 	software_interrupt_mask_signal(SIGUSR1);
 	software_interrupt_mask_signal(SIGALRM);
-	signal(SIGPIPE, SIG_IGN);
 
 	http_parser_settings_initialize();
 	admissions_control_initialize();

runtime/src/software_interrupt.c

@@ -10,6 +10,7 @@
 #include "arch/context.h"
 #include "current_sandbox.h"
 #include "debuglog.h"
+#include "global_request_scheduler.h"
 #include "local_runqueue.h"
 #include "module.h"
 #include "panic.h"
@@ -28,7 +29,8 @@ uint64_t         software_interrupt_interval_duration_in_cycles;
  * Thread Globals *
  *****************/
 
-__thread static volatile sig_atomic_t software_interrupt_SIGALRM_count = 0;
+__thread static volatile sig_atomic_t software_interrupt_SIGALRM_kernel_count = 0;
+__thread static volatile sig_atomic_t software_interrupt_SIGALRM_thread_count = 0;
 __thread static volatile sig_atomic_t software_interrupt_SIGUSR_count         = 0;
 __thread volatile sig_atomic_t        software_interrupt_is_disabled          = 0;
 
@@ -54,15 +56,33 @@ sigalrm_propagate_workers(siginfo_t *signal_info)
 {
 	/* Signal was sent directly by the kernel, so forward to other threads */
 	if (signal_info->si_code == SI_KERNEL) {
+		software_interrupt_SIGALRM_kernel_count++;
+#ifdef LOG_PREEMPTION
+		debuglog("Kernel SIGALRM: %d!\n", software_interrupt_SIGALRM_kernel_count);
+#endif
 		for (int i = 0; i < runtime_worker_threads_count; i++) {
 			if (pthread_self() == runtime_worker_threads[i]) continue;
 
 			/* All threads should have been initialized */
 			assert(runtime_worker_threads[i] != 0);
 
+			/* If using EDF, conditionally send signals. If not, broadcast */
+			if (runtime_sigalrm_handler == RUNTIME_SIGALRM_HANDLER_TRIAGED) {
+				uint64_t local_deadline  = runtime_worker_threads_deadline[i];
+				uint64_t global_deadline = global_request_scheduler_peek();
+				if (global_deadline < local_deadline) pthread_kill(runtime_worker_threads[i], SIGALRM);
+				return;
+			} else if (runtime_sigalrm_handler == RUNTIME_SIGALRM_HANDLER_BROADCAST) {
 				pthread_kill(runtime_worker_threads[i], SIGALRM);
+			} else {
+				panic("Unexpected SIGALRM Handler: %d\n", runtime_sigalrm_handler)
+			}
 		}
 	} else {
+		software_interrupt_SIGALRM_thread_count++;
+#ifdef LOG_PREEMPTION
+		debuglog("Thread SIGALRM: %d!\n", software_interrupt_SIGALRM_thread_count);
+#endif
 		/* Signal forwarded from another thread. Just confirm it resulted from pthread_kill */
 		assert(signal_info->si_code == SI_TKILL);
 	}
@@ -79,7 +99,6 @@ sigalrm_handler(siginfo_t *signal_info, ucontext_t *user_context, struct sandbox
 {
 	sigalrm_propagate_workers(signal_info);
 
-	software_interrupt_SIGALRM_count++;
 
 	/* NOOP if software interrupts not enabled */
 	if (!software_interrupt_is_enabled()) return;
@@ -150,7 +169,7 @@ static inline void
 software_interrupt_validate_worker()
 {
 #ifndef NDEBUG
-	if (!runtime_is_worker()) panic("A non-worker thread received has unexpectedly received a signal!");
+	if (!runtime_is_worker()) panic("A non-worker thread has unexpectedly received a signal!");
 #endif
 }
 

runtime/src/worker_thread.c

@@ -33,6 +33,9 @@ __thread uint64_t worker_thread_lock_duration;
 /* Timestamp when worker thread began executing */
 __thread uint64_t worker_thread_start_timestamp;
 
+/* Used to index into global arguments and deadlines arrays */
+__thread int worker_thread_idx;
+
 /***********************
  * Worker Thread Logic *
  **********************/
@@ -97,6 +100,9 @@ worker_thread_switch_to_sandbox(struct sandbox *next_sandbox)
 	/* Get the old sandbox we're switching from */
 	struct sandbox *current_sandbox = current_sandbox_get();
 
+	/* Update the worker's absolute deadline */
+	runtime_worker_threads_deadline[worker_thread_idx] = next_sandbox->absolute_deadline;
+
 	if (current_sandbox == NULL) {
 		/* Switching from "Base Context" */
 
@@ -171,6 +177,7 @@ worker_thread_switch_to_base_context()
 	worker_thread_transition_exiting_sandbox(current_sandbox);
 	current_sandbox_set(NULL);
 	assert(worker_thread_base_context.variant == ARCH_CONTEXT_VARIANT_FAST);
+	runtime_worker_threads_deadline[worker_thread_idx] = UINT64_MAX;
 	arch_context_switch(current_context, &worker_thread_base_context);
 	software_interrupt_enable();
 }
@@ -283,11 +290,14 @@ worker_thread_execute_epoll_loop(void)
 /**
  * The entry function for sandbox worker threads
  * Initializes thread-local state, unmasks signals, sets up epoll loop and
- * @param return_code - argument provided by pthread API. We set to -1 on error
+ * @param argument - argument provided by pthread API. We set to -1 on error
  */
 void *
-worker_thread_main(void *return_code)
+worker_thread_main(void *argument)
 {
+	/* Index was passed via argument */
+	worker_thread_idx = *(int *)argument;
+
 	/* Initialize Bookkeeping */
 	worker_thread_start_timestamp = __getcycles();
 	worker_thread_lock_duration   = 0;
@@ -322,8 +332,6 @@ worker_thread_main(void *return_code)
 		software_interrupt_unmask_signal(SIGUSR1);
 	}
 
-	signal(SIGPIPE, SIG_IGN);
-
 	/* Initialize epoll */
 	worker_thread_epoll_file_descriptor = epoll_create1(0);
 	if (unlikely(worker_thread_epoll_file_descriptor < 0)) panic_err();