add a flag to exit while loop for each worker thread

install_deb.sh

@@ -10,7 +10,6 @@ if [[ $ARCH = "x86_64" ]]; then
 elif [[ $ARCH = "aarch64" ]]; then
 	SHFMT_URL=https://github.com/patrickvane/shfmt/releases/download/master/shfmt_linux_arm
 	echo "ARM64 support is still a work in progress!"
-	exit 1
 else
 	echo "This script only supports x86_64 and aarch64"
 	exit 1
@@ -64,8 +63,6 @@ wget $SHFMT_URL -O shfmt && chmod +x shfmt && sudo mv shfmt /usr/local/bin/shfmt
 
 sudo ./install_llvm.sh $LLVM_VERSION
 
-curl -sS -L -O $WASI_SDK_URL && sudo dpkg -i wasi-sdk_12.0_amd64.deb && rm -f wasi-sdk_12.0_amd64.deb
-
 if [ -z "${WASI_SDK_PATH}" ]; then
 	export WASI_SDK_PATH=/opt/wasi-sdk
 	echo "export WASI_SDK_PATH=/opt/wasi-sdk" >> ~/.bashrc

runtime/Makefile

@@ -94,7 +94,7 @@ BINARY_NAME=sledgert
 
 # This flag tracks the total number of sandboxes in the various states
 # It is useful to debug if sandboxes are "getting caught" in a particular state
-# CFLAGS += -DSANDBOX_STATE_TOTALS
+ CFLAGS += -DSANDBOX_STATE_TOTALS
 
 # This flag enables an per-worker atomic count of sandbox's local runqueue count in thread local storage
 # Useful to debug if sandboxes are "getting caught" or "leaking" while in a local runqueue

runtime/include/arch/aarch64/context.h

@@ -4,7 +4,6 @@
 #include <assert.h>
 
 #include "arch/common.h"
-#include "current_sandbox.h"
 
 #define ARCH_SIG_JMP_OFF 0x100 /* Based on code generated! */
 
@@ -47,24 +46,20 @@ arch_context_switch(struct arch_context *a, struct arch_context *b)
 	 * Assumption: In the case of a slow context switch, the caller
 	 * set current_sandbox to the sandbox containing the target context
 	 */
-	if (b->variant == ARCH_CONTEXT_VARIANT_SLOW) {
+	/*if (b->variant == ARCH_CONTEXT_VARIANT_SLOW) {
 		struct sandbox *current = current_sandbox_get();
 		assert(current != NULL && b == &current->ctxt);
-	}
+	}*/
 #endif
 
 	/* if both a and b are NULL, there is no state change */
-	assert(a != NULL || b != NULL);
+	assert(a != NULL && b != NULL);
 
 	/* Assumption: The caller does not switch to itself */
 	assert(a != b);
 
-	/* Set any NULLs to worker_thread_base_context to resume execution of main */
-	if (a == NULL) a = &worker_thread_base_context;
-	if (b == NULL) b = &worker_thread_base_context;
-
 	/* A Transition {Unused, Running} -> Fast */
-	assert(a->variant == ARCH_CONTEXT_VARIANT_UNUSED || a->variant == ARCH_CONTEXT_VARIANT_RUNNING);
+	assert(a->variant == ARCH_CONTEXT_VARIANT_RUNNING);
 
 	/* B Transition {Fast, Slow} -> Running */
 	assert(b->variant == ARCH_CONTEXT_VARIANT_FAST || b->variant == ARCH_CONTEXT_VARIANT_SLOW);
@@ -87,6 +82,8 @@ arch_context_switch(struct arch_context *a, struct arch_context *b)
 	                 "ldr x1, [%[bv]]\n\t"
 	                 "sub x1, x1, #2\n\t"
 	                 "cbz x1, slow%=\n\t"
+			 "mov x3, #3\n\t"
+			 "str x3, [%[bv]]\n\t" /* b->variant = ARCH_CONTEXT_VARIANT_RUNNING; */
 	                 "ldr x0, [%[b]]\n\t"
 	                 "ldr x1, [%[b], 8]\n\t"
 	                 "mov sp, x0\n\t"
@@ -95,8 +92,6 @@ arch_context_switch(struct arch_context *a, struct arch_context *b)
 	                 "br %[slowpath]\n\t"
 	                 ".align 8\n\t"
 	                 "reset%=:\n\t"
-	                 "mov x1, #3\n\t"
-	                 "str x1, [%[bv]]\n\t"
 	                 ".align 8\n\t"
 	                 "exit%=:\n\t"
 	                 :
@@ -109,6 +104,37 @@ arch_context_switch(struct arch_context *a, struct arch_context *b)
 	return 0;
 }
 
+
+/**
+ * Load a new sandbox that preempted an existing sandbox, restoring only the
+ * instruction pointer and stack pointer registers.
+ * @param active_context - the context of the current worker thread
+ * @param sandbox_context - the context that we want to restore
+ */
+static inline void
+arch_context_restore_fast(mcontext_t *active_context, struct arch_context *sandbox_context)
+{
+        assert(active_context != NULL);
+        assert(sandbox_context != NULL);
+
+        /* Assumption: Base Context is only ever used by arch_context_switch */
+        assert(sandbox_context != &worker_thread_base_context);
+
+        assert(sandbox_context->regs[UREG_SP]);
+        assert(sandbox_context->regs[UREG_IP]);
+
+        /* Transitioning from Fast -> Running */
+        assert(sandbox_context->variant == ARCH_CONTEXT_VARIANT_FAST);
+        sandbox_context->variant = ARCH_CONTEXT_VARIANT_RUNNING;
+
+        //active_context->gregs[REG_RSP] = sandbox_context->regs[UREG_SP];
+        //active_context->gregs[REG_RIP] = sandbox_context->regs[UREG_IP];
+	active_context->sp = sandbox_context->regs[UREG_SP];
+        active_context->pc = sandbox_context->regs[UREG_IP];
+}
+
 #else
 #warning "Neither AARCH64 nor aarch64 was defined, but aarch64/context.h was included!"
 #endif
+
+

runtime/include/auto_buf.h

@@ -10,6 +10,12 @@ struct auto_buf {
 	size_t size;
 };
 
+static inline void auto_buf_copy(struct auto_buf *dest, struct auto_buf *source) {
+	if (dest == NULL || source == NULL) return;
+	fwrite(source->data, 1, source->size, dest->handle);
+	fflush(dest->handle);
+}
+
 static inline int
 auto_buf_init(struct auto_buf *buf)
 {

runtime/include/http_request.h

@@ -12,11 +12,28 @@ struct http_header {
 	int   value_length;
 };
 
+static inline void http_header_copy(struct http_header *dest, struct http_header *source) {
+	if (source == NULL || dest == NULL) return;
+
+	dest->key = (char*) malloc (source->key_length);	
+	memcpy(dest->key, source->key, source->key_length);
+	dest->key_length = source->key_length;
+
+	dest->value = (char*) malloc (source->value_length);
+	memcpy(dest->value, source->value, source->value_length);
+	dest->value_length = source->value_length;
+}
+
 struct http_query_param {
 	char value[HTTP_MAX_QUERY_PARAM_LENGTH];
 	int  value_length;
 };
 
+static inline void http_query_param_copy(struct http_query_param *dest, struct http_query_param *source) {
+	if (source == NULL || dest == NULL) return;
+	memcpy(dest->value, source->value, HTTP_MAX_QUERY_PARAM_LENGTH);
+	dest->value_length = source->value_length;
+}
 struct http_request {
 	char                    full_url[HTTP_MAX_FULL_URL_LENGTH];
 	struct http_header      headers[HTTP_MAX_HEADER_COUNT];
@@ -40,4 +57,29 @@ struct http_request {
 	int cursor; /* Sandbox cursor (offset from body pointer) */
 };
 
+static inline void http_request_copy(struct http_request *dest, struct http_request *source) {
+
+	if (dest == NULL || source == NULL) return;
+	memcpy(dest->full_url, source->full_url, HTTP_MAX_FULL_URL_LENGTH);
+	for (int i = 0; i < HTTP_MAX_HEADER_COUNT; i++) {
+		http_header_copy(&(dest->headers[i]), &(source->headers[i]));
+	}
+	dest->header_count = source->header_count;
+	dest->method = source->method;
+
+	for (int i = 0; i < HTTP_MAX_QUERY_PARAM_COUNT; i++) {
+		http_query_param_copy(&(dest->query_params[i]), &(source->query_params[i]));
+	}
+	dest->query_params_count = source->query_params_count;
+	dest->body = (char*) malloc (source->body_length);
+	memcpy(dest->body, source->body, source->body_length);
+	dest->body_length = source->body_length;
+	dest->body_length_read = source->body_length_read;
+	dest->length_parsed = source->length_parsed;
+	dest->last_was_value = source->last_was_value;
+	dest->header_end = source->header_end;
+	dest->message_begin = source->message_begin;
+	dest->message_end = source->message_end;
+
+}
 void http_request_print(struct http_request *http_request);

runtime/include/http_session.h

@@ -22,6 +22,7 @@
 #include "tcp_session.h"
 #include "tenant.h"
 
+
 enum http_session_state
 {
 	HTTP_SESSION_UNINITIALIZED = 0,
@@ -59,8 +60,22 @@ struct http_session {
 	uint64_t                response_sent_timestamp;
 };
 
+static inline void http_session_copy(struct http_session *dest, struct http_session *source) {
+	if (dest == NULL || source == NULL) return;
+
+	dest->tag = source->tag;
+	http_request_copy(&dest->http_request, &source->http_request);
+	auto_buf_copy(&dest->request_buffer, &source->request_buffer);
+	auto_buf_copy(&dest->response_header, &source->response_header);
+	dest->response_header_written = source->response_header_written;
+	auto_buf_copy(&dest->response_body, &source->response_body);
+	dest->response_body_written = source->response_body_written;
+	dest->route = source->route;
+
+}
 extern void http_session_perf_log_print_entry(struct http_session *http_session);
 
+
 /**
  * Initalize state associated with an http parser
  * Because the http_parser structure uses pointers to the request buffer, if realloc moves the request
@@ -102,6 +117,7 @@ http_session_init(struct http_session *session, int socket_descriptor, const str
 
 	/* Defer initializing response_body until we've matched a route */
 	auto_buf_init(&session->response_header);
+	
 
 	session->state = HTTP_SESSION_INITIALIZED;
 
@@ -112,14 +128,23 @@ static inline int
 http_session_init_response_body(struct http_session *session)
 {
 	assert(session != NULL);
-	assert(session->response_body.data == NULL);
-	assert(session->response_body.size == 0);
-	assert(session->response_body_written == 0);
-
-	int rc = auto_buf_init(&session->response_body);
-	if (rc < 0) {
-		auto_buf_deinit(&session->request_buffer);
-		return -1;
+	//assert(session->response_body.data == NULL);
+	//assert(session->response_body.size == 0);
+	//assert(session->response_body_written == 0);
+
+	//int rc = auto_buf_init(&session->response_body);
+	//if (rc < 0) {
+	//	auto_buf_deinit(&session->request_buffer);
+	//	return -1;
+	//}
+	session->response_body.size = 0;
+	session->response_body_written = 0;
+	if (session->response_body.data == NULL) {
+		int rc = auto_buf_init(&session->response_body);
+		if (rc < 0) {
+			auto_buf_deinit(&session->request_buffer);
+			return -1;
+		}
 	}
 
 	return 0;
@@ -153,9 +178,9 @@ http_session_deinit(struct http_session *session)
 {
 	assert(session);
 
-	auto_buf_deinit(&session->request_buffer);
-	auto_buf_deinit(&session->response_header);
-	auto_buf_deinit(&session->response_body);
+	//auto_buf_deinit(&session->request_buffer);
+	//auto_buf_deinit(&session->response_header);
+	//auto_buf_deinit(&session->response_body);
 }
 
 static inline void
@@ -163,8 +188,8 @@ http_session_free(struct http_session *session)
 {
 	assert(session);
 
-	http_session_deinit(session);
-	free(session);
+	//http_session_deinit(session);
+	//free(session);
 }
 
 /**
@@ -265,7 +290,7 @@ http_session_send_response_body(struct http_session *session, void_star_cb on_ea
 
 	/* Assumption: Already flushed in order to write content-length to header */
 	// TODO: Test if body is empty
-
+        printf("body is %s\n", session->response_body.data);
 	while (session->response_body_written < session->response_body.size) {
 		ssize_t sent =
 		  tcp_session_send(session->socket,
@@ -384,7 +409,6 @@ http_session_receive_request(struct http_session *session, void_star_cb on_eagai
 		if (old_buffer != session->request_buffer.data) {
 			http_request->body = header_length ? session->request_buffer.data + header_length : NULL;
 		}
-
 		if (http_session_parse(session, bytes_received) == -1) goto err;
 	}
 

runtime/include/lock.h

@@ -7,6 +7,9 @@
 #include "arch/getcycles.h"
 #include "runtime.h"
 
+extern uint64_t total_held[1024];
+extern uint64_t longest_held[1024];
+extern thread_local int thread_id;
 
 /* A linked list of nodes */
 struct lock_wrapper {
@@ -73,9 +76,13 @@ lock_unlock(lock_t *self, lock_node_t *node)
 
 	ck_spinlock_mcs_unlock(&self->lock, &node->node);
 	uint64_t now = __getcycles();
-	assert(node->time_locked < now);
+	assert(node->time_locked <= now);
 	uint64_t duration = now - node->time_locked;
 	node->time_locked = 0;
 	if (unlikely(duration > self->longest_held)) { self->longest_held = duration; }
 	self->total_held += duration;
+	if (unlikely(duration > longest_held[thread_id])) {
+		longest_held[thread_id] = duration;
+	}
+	total_held[thread_id] += duration;
 }

runtime/include/sandbox_perf_log.h

@@ -48,6 +48,7 @@ sandbox_perf_log_print_entry(struct sandbox *sandbox)
 	        runtime_processor_speed_MHz);
 }
 
+
 static inline void
 sandbox_perf_log_init()
 {

runtime/include/sandbox_set_as_error.h

@@ -60,7 +60,7 @@ sandbox_set_as_error(struct sandbox *sandbox, sandbox_state_t last_state)
 	/* Return HTTP session to listener core to be written back to client */
 	http_session_set_response_header(sandbox->http, 500);
 	sandbox->http->state = HTTP_SESSION_EXECUTION_COMPLETE;
-	http_session_send_response(sandbox->http, (void_star_cb)listener_thread_register_http_session);
+	//http_session_send_response(sandbox->http, (void_star_cb)listener_thread_register_http_session);
 	sandbox->http = NULL;
 
 	/* Terminal State Logging */

runtime/include/sandbox_set_as_initialized.h

@@ -34,7 +34,7 @@ sandbox_set_as_initialized(struct sandbox *sandbox, sandbox_state_t last_state)
 	}
 
 	/* State Change Bookkeeping */
-	assert(now > sandbox->timestamp_of.last_state_change);
+	assert(now >= sandbox->timestamp_of.last_state_change);
 	sandbox->last_state_duration = now - sandbox->timestamp_of.last_state_change;
 	sandbox->duration_of_state[last_state] += sandbox->last_state_duration;
 	sandbox->timestamp_of.last_state_change = now;

runtime/include/sandbox_set_as_returned.h

@@ -43,7 +43,7 @@ sandbox_set_as_returned(struct sandbox *sandbox, sandbox_state_t last_state)
 	}
 
 	/* State Change Bookkeeping */
-	assert(now > sandbox->timestamp_of.last_state_change);
+	assert(now >= sandbox->timestamp_of.last_state_change);
 	sandbox->last_state_duration = now - sandbox->timestamp_of.last_state_change;
 	sandbox->duration_of_state[last_state] += sandbox->last_state_duration;
 	sandbox->timestamp_of.last_state_change = now;
@@ -51,9 +51,9 @@ sandbox_set_as_returned(struct sandbox *sandbox, sandbox_state_t last_state)
 	sandbox_state_totals_increment(SANDBOX_RETURNED);
 	sandbox_state_totals_decrement(last_state);
 
-	http_session_set_response_header(sandbox->http, 200);
+	//http_session_set_response_header(sandbox->http, 200);
 	sandbox->http->state = HTTP_SESSION_EXECUTION_COMPLETE;
-	http_session_send_response(sandbox->http, (void_star_cb)listener_thread_register_http_session);
+	//http_session_send_response(sandbox->http, (void_star_cb)listener_thread_register_http_session);
 	sandbox->http = NULL;
 
 	/* State Change Hooks */

runtime/include/sandbox_set_as_running_sys.h

@@ -39,7 +39,7 @@ sandbox_set_as_running_sys(struct sandbox *sandbox, sandbox_state_t last_state)
 	}
 
 	/* State Change Bookkeeping */
-	assert(now > sandbox->timestamp_of.last_state_change);
+	assert(now >= sandbox->timestamp_of.last_state_change);
 	sandbox->last_state_duration = now - sandbox->timestamp_of.last_state_change;
 	sandbox->duration_of_state[last_state] += sandbox->last_state_duration;
 	sandbox->timestamp_of.last_state_change = now;

runtime/include/sandbox_set_as_running_user.h

@@ -35,7 +35,7 @@ sandbox_set_as_running_user(struct sandbox *sandbox, sandbox_state_t last_state)
 
 
 	/* State Change Bookkeeping */
-	assert(now > sandbox->timestamp_of.last_state_change);
+	assert(now >= sandbox->timestamp_of.last_state_change);
 	sandbox->last_state_duration = now - sandbox->timestamp_of.last_state_change;
 	sandbox->duration_of_state[last_state] += sandbox->last_state_duration;
 	sandbox->timestamp_of.last_state_change = now;

runtime/include/sandbox_types.h

@@ -46,6 +46,8 @@ struct sandbox {
 
 	/* HTTP State */
 	struct http_session *http;
+	struct auto_buf response_body;
+	struct auto_buf response_header;
 
 	/* WebAssembly Module State */
 	struct module *module; /* the module this is an instance of */
@@ -66,6 +68,8 @@ struct sandbox {
 	                                 deadline (cycles) */
 	uint64_t total_time;          /* Total time from Request to Response */
 
+	int cursor;
+
 	/* System Interface State */
 	int32_t         return_value;
 	wasi_context_t *wasi_context;

runtime/include/scheduler.h

@@ -24,6 +24,7 @@
 #include "sandbox_set_as_running_user.h"
 #include "scheduler_options.h"
 
+extern thread_local bool pthread_stop;
 
 /**
  * This scheduler provides for cooperative and preemptive multitasking in a OS process's userspace.
@@ -392,7 +393,7 @@ scheduler_switch_to_base_context(struct arch_context *current_context)
 static inline void
 scheduler_idle_loop()
 {
-	while (true) {
+	while (!pthread_stop) {
 		/* Assumption: only called by the "base context" */
 		assert(current_sandbox_get() == NULL);
 

runtime/include/software_interrupt.h

@@ -31,7 +31,7 @@ software_interrupt_mask_signal(int signal)
 	sigset_t set;
 	int      return_code;
 
-	assert(signal == SIGALRM || signal == SIGUSR1 || signal == SIGFPE || signal == SIGSEGV);
+	assert(signal == SIGALRM || signal == SIGUSR1 || signal == SIGFPE || signal == SIGSEGV || signal == SIGINT);
 	/* all threads created by the calling thread will have signal blocked */
 	sigemptyset(&set);
 	sigaddset(&set, signal);
@@ -55,7 +55,7 @@ software_interrupt_unmask_signal(int signal)
 	sigset_t set;
 	int      return_code;
 
-	assert(signal == SIGALRM || signal == SIGUSR1 || signal == SIGFPE || signal == SIGSEGV);
+	assert(signal == SIGALRM || signal == SIGUSR1 || signal == SIGFPE || signal == SIGSEGV || signal == SIGINT);
 	/* all threads created by the calling thread will have signal unblocked */
 	sigemptyset(&set);
 	sigaddset(&set, signal);

runtime/src/current_sandbox.c

@@ -11,9 +11,15 @@
 #include "sandbox_set_as_running_user.h"
 #include "sandbox_set_as_running_sys.h"
 #include "scheduler.h"
+#include "http_session.h"
 #include "software_interrupt.h"
 #include "wasi.h"
 
+extern struct http_session *g_session;
+extern struct tenant *g_tenant;
+extern int g_client_socket;
+extern struct sockaddr *g_client_address;
+
 thread_local struct sandbox *worker_thread_current_sandbox = NULL;
 
 /**
@@ -63,6 +69,34 @@ current_sandbox_exit()
 		      sandbox_state_stringify(exiting_sandbox->state));
 	}
 
+	/* generate a new request and enqueue it to the global queue */
+	assert(g_tenant != NULL);
+	struct tenant *tenant = g_tenant;
+
+	uint64_t request_arrival_timestamp = __getcycles();
+	http_total_increment_request();
+
+	/* Allocate http session */
+	//struct http_session *session = http_session_alloc(g_client_socket, (const struct sockaddr *)&g_client_address, tenant, request_arrival_timestamp);
+
+	//assert(session != NULL);
+	//http_session_copy(session, g_session);
+	struct http_session *session = g_session;
+	assert(session->route != NULL);
+	struct sandbox *sandbox = sandbox_alloc(session->route->module, session, session->route, session->tenant, 1);
+        if (unlikely(sandbox == NULL)) {
+                printf("Failed to allocate sandbox\n"); 
+                exit(-1);
+        }
+
+        /* If the global request scheduler is full, return a 429 to the client */
+        if (unlikely(global_request_scheduler_add(sandbox) == NULL)) {
+                printf("Failed to add sandbox to global queue\n");
+		exit(-1);
+                
+        }
+
+	/****************************end**************************/
 	scheduler_cooperative_sched(true);
 
 	/* The scheduler should never switch back to completed sandboxes */

runtime/src/global_request_scheduler_minheap.c

@@ -19,7 +19,7 @@ global_request_scheduler_minheap_add(struct sandbox *sandbox)
 {
 	assert(sandbox);
 	assert(global_request_scheduler_minheap);
-	if (unlikely(!listener_thread_is_running())) panic("%s is only callable by the listener thread\n", __func__);
+	//if (unlikely(!listener_thread_is_running())) panic("%s is only callable by the listener thread\n", __func__);
 
 	int return_code = priority_queue_enqueue(global_request_scheduler_minheap, sandbox);
 
@@ -75,7 +75,7 @@ sandbox_get_priority_fn(void *element)
 void
 global_request_scheduler_minheap_initialize()
 {
-	global_request_scheduler_minheap = priority_queue_initialize(4096, true, sandbox_get_priority_fn);
+	global_request_scheduler_minheap = priority_queue_initialize(4096000, true, sandbox_get_priority_fn);
 
 	struct global_request_scheduler_config config = {
 		.add_fn               = global_request_scheduler_minheap_add,

runtime/src/libc/wasi_impl_serverless.c

@@ -660,7 +660,7 @@ wasi_snapshot_preview1_backing_fd_read(wasi_context_t *context, __wasi_fd_t fd,
 	if (fd == STDIN_FILENO) {
 		struct sandbox      *current_sandbox = current_sandbox_get();
 		struct http_request *current_request = &current_sandbox->http->http_request;
-		int                  old_read        = current_request->cursor;
+		int                  old_read        = current_sandbox->cursor;
 		int                  bytes_to_read   = current_request->body_length - old_read;
 
 		assert(current_request->body_length >= 0);
@@ -669,13 +669,13 @@ wasi_snapshot_preview1_backing_fd_read(wasi_context_t *context, __wasi_fd_t fd,
 			if (bytes_to_read == 0) goto done;
 
 			int amount_to_copy = iovs[i].buf_len > bytes_to_read ? bytes_to_read : iovs[i].buf_len;
-			memcpy(iovs[i].buf, current_request->body + current_request->cursor, amount_to_copy);
-			current_request->cursor += amount_to_copy;
-			bytes_to_read = current_request->body_length - current_request->cursor;
+			memcpy(iovs[i].buf, current_request->body + current_sandbox->cursor, amount_to_copy);
+			current_sandbox->cursor += amount_to_copy;
+			bytes_to_read = current_request->body_length - current_sandbox->cursor;
 		}
 
 	done:
-		*nwritten_retptr = current_request->cursor - old_read;
+		*nwritten_retptr = current_sandbox->cursor - old_read;
 		return __WASI_ERRNO_SUCCESS;
 	}
 
@@ -796,7 +796,8 @@ wasi_snapshot_preview1_backing_fd_write(wasi_context_t *context, __wasi_fd_t fd,
 				debuglog("STDERR from Sandbox: %.*s", iovs[i].buf_len, iovs[i].buf);
 			}
 #endif
-			rc = fwrite(iovs[i].buf, 1, iovs[i].buf_len, s->http->response_body.handle);
+			//rc = fwrite(iovs[i].buf, 1, iovs[i].buf_len, s->http->response_body.handle);
+			rc = fwrite(iovs[i].buf, 1, iovs[i].buf_len, s->response_body.handle);
 			if (rc != iovs[i].buf_len) return __WASI_ERRNO_FBIG;
 
 			nwritten += rc;

runtime/src/listener_thread.c

@@ -13,6 +13,17 @@
 #include "tenant_functions.h"
 #include "http_session_perf_log.h"
 
+extern thread_local int thread_id;
+extern void http_session_copy(struct http_session *dest, struct http_session *source);
+/////////////////xiaosu for test//////////////////
+extern bool first_request_comming;
+time_t t_start;
+struct http_session *g_session = NULL;
+struct tenant *g_tenant = NULL;
+int g_client_socket = -1;
+struct sockaddr *g_client_address = NULL;
+////////////////xiaosu end for test///////////////
+
 static void listener_thread_unregister_http_session(struct http_session *http);
 static void panic_on_epoll_error(struct epoll_event *evt);
 
@@ -168,6 +179,14 @@ panic_on_epoll_error(struct epoll_event *evt)
 static void
 on_client_request_arrival(int client_socket, const struct sockaddr *client_address, struct tenant *tenant)
 {
+	if (g_client_socket == -1) {
+		g_client_socket = client_socket;
+	}
+
+	if (g_client_address == NULL) {
+		g_client_address = client_address;
+	}
+
 	uint64_t request_arrival_timestamp = __getcycles();
 
 	http_total_increment_request();
@@ -191,6 +210,11 @@ on_client_request_arrival(int client_socket, const struct sockaddr *client_addre
 static void
 on_client_request_receiving(struct http_session *session)
 {
+	if (first_request_comming == false){
+               t_start = time(NULL);
+               first_request_comming = true;
+        }
+
 	/* Read HTTP request */
 	int rc = http_session_receive_request(session, (void_star_cb)listener_thread_register_http_session);
 	if (likely(rc == 0)) {
@@ -242,6 +266,13 @@ on_client_request_received(struct http_session *session)
 
 	/* Allocate a Sandbox */
 	session->state          = HTTP_SESSION_EXECUTING;
+	if (g_session == NULL) {
+		/* Allocate HTTP Session */
+        	g_session = http_session_alloc(session->socket, (const struct sockaddr *)&(session->client_address),
+                                                          session->tenant, session->request_arrival_timestamp);
+
+		http_session_copy(g_session, session);
+	}
 	struct sandbox *sandbox = sandbox_alloc(route->module, session, route, session->tenant, work_admitted);
 	if (unlikely(sandbox == NULL)) {
 		debuglog("Failed to allocate sandbox\n");
@@ -317,6 +348,9 @@ on_tenant_socket_epoll_event(struct epoll_event *evt)
 
 	struct tenant *tenant = evt->data.ptr;
 	assert(tenant);
+	if (g_tenant == NULL) {
+		g_tenant = tenant;
+	}
 
 	/* Accept Client Request as a nonblocking socket, saving address information */
 	struct sockaddr_in client_address;
@@ -401,6 +435,7 @@ on_client_socket_epoll_event(struct epoll_event *evt)
 noreturn void *
 listener_thread_main(void *dummy)
 {
+	thread_id = 200;
 	struct epoll_event epoll_events[RUNTIME_MAX_EPOLL_EVENTS];
 
 	metrics_server_init();

runtime/src/local_runqueue.c

@@ -6,6 +6,7 @@
 
 #include "local_runqueue.h"
 
+thread_local uint32_t total_local_requests = 0;
 static struct local_runqueue_config local_runqueue;
 
 #ifdef LOG_LOCAL_RUNQUEUE
@@ -27,6 +28,7 @@ void
 local_runqueue_add(struct sandbox *sandbox)
 {
 	assert(local_runqueue.add_fn != NULL);
+	total_local_requests++;
 #ifdef LOG_LOCAL_RUNQUEUE
 	local_runqueue_count++;
 #endif

runtime/src/local_runqueue_minheap.c

@@ -12,6 +12,8 @@
 #include "sandbox_functions.h"
 #include "runtime.h"
 
+uint64_t total_held[1024] = {0};
+uint64_t longest_held[1024] = {0};
 thread_local static struct priority_queue *local_runqueue_minheap;
 
 /**

runtime/src/main.c

@@ -35,6 +35,8 @@ uint32_t runtime_first_worker_processor  = 1;
 uint32_t runtime_processor_speed_MHz     = 0;
 uint32_t runtime_total_online_processors = 0;
 uint32_t runtime_worker_threads_count    = 0;
+bool first_request_comming 		 = false;
+thread_local int thread_id 		 = -1;
 
 enum RUNTIME_SIGALRM_HANDLER runtime_sigalrm_handler = RUNTIME_SIGALRM_HANDLER_BROADCAST;
 
@@ -67,15 +69,20 @@ runtime_allocate_available_cores()
 
 	pretty_print_key_value("Core Count (Online)", "%u\n", runtime_total_online_processors);
 
-	/* If more than two cores are available, leave core 0 free to run OS tasks */
-	if (runtime_total_online_processors > 2) {
-		runtime_first_worker_processor = 2;
-		max_possible_workers           = runtime_total_online_processors - 2;
-	} else if (runtime_total_online_processors == 2) {
-		runtime_first_worker_processor = 1;
-		max_possible_workers           = runtime_total_online_processors - 1;
+	char* first_core_id = getenv("SLEDGE_FIRST_COREID");
+	if (first_core_id != NULL) {
+		runtime_first_worker_processor = atoi(first_core_id);
 	} else {
-		panic("Runtime requires at least two cores!");
+		/* If more than two cores are available, leave core 0 free to run OS tasks */
+		if (runtime_total_online_processors > 2) {
+                	runtime_first_worker_processor = 2;
+                        max_possible_workers           = runtime_total_online_processors - 2;
+                } else if (runtime_total_online_processors == 2) {
+                        runtime_first_worker_processor = 1;
+                        max_possible_workers           = runtime_total_online_processors - 1;
+                } else {
+			panic("Runtime requires at least two cores!");
+                }
 	}
 
 
@@ -476,6 +483,8 @@ main(int argc, char **argv)
 
 	printf("Runtime Environment:\n");
 
+	runtime_processor_speed_MHz = 1500;
+
 	runtime_set_resource_limits_to_max();
 	runtime_allocate_available_cores();
 	runtime_configure();
@@ -484,7 +493,7 @@ main(int argc, char **argv)
 
 	listener_thread_initialize();
 	runtime_start_runtime_worker_threads();
-	runtime_get_processor_speed_MHz();
+	//runtime_get_processor_speed_MHz();
 	runtime_configure_worker_spinloop_pause();
 	software_interrupt_arm_timer();
 

runtime/src/sandbox.c

@@ -25,6 +25,21 @@ sandbox_log_allocation(struct sandbox *sandbox)
 #endif
 }
 
+int
+init_response_body(struct sandbox *sandbox)
+{
+        if (sandbox->response_body.data == NULL) {
+		sandbox->response_body.size = 0;
+                int rc = auto_buf_init(&sandbox->response_body);
+                if (rc < 0) {
+                        printf("failed to init http body autobuf, exit\n");
+                        exit(-1);
+                }
+        }
+
+        return 0;
+}
+
 /**
  * Allocates a WebAssembly linear memory for a sandbox based on the starting_pages and max_pages globals present in
  * the associated *.so module
@@ -94,7 +109,8 @@ sandbox_prepare_execution_environment(struct sandbox *sandbox)
 
 	int rc;
 
-	rc = http_session_init_response_body(sandbox->http);
+	//rc = http_session_init_response_body(sandbox->http);
+	rc = init_response_body(sandbox);
 	if (rc < 0) {
 		error_message = "failed to allocate response body";
 		goto err_globals_allocation_failed;
@@ -135,6 +151,14 @@ err_http_allocation_failed:
 	goto done;
 }
 
+void
+deinit_response_body(struct sandbox *sandbox)
+{
+        auto_buf_deinit(&sandbox->response_body);
+}
+
+
+
 void
 sandbox_init(struct sandbox *sandbox, struct module *module, struct http_session *session, struct route *route,
              struct tenant *tenant, uint64_t admissions_estimate)
@@ -148,7 +172,7 @@ sandbox_init(struct sandbox *sandbox, struct module *module, struct http_session
 	ps_list_init_d(sandbox);
 
 	/* Allocate HTTP session structure */
-	assert(session);
+	//assert(session);
 	sandbox->http   = session;
 	sandbox->tenant = tenant;
 	sandbox->route  = route;
@@ -211,6 +235,8 @@ sandbox_deinit(struct sandbox *sandbox)
 	/* Linear Memory and Guard Page should already have been munmaped and set to NULL */
 	assert(sandbox->memory == NULL);
 
+	deinit_response_body(sandbox);
+
 	if (likely(sandbox->stack != NULL)) sandbox_free_stack(sandbox);
 	if (likely(sandbox->globals.buffer != NULL)) sandbox_free_globals(sandbox);
 	if (likely(sandbox->wasi_context != NULL)) wasi_context_destroy(sandbox->wasi_context);

runtime/src/software_interrupt.c

@@ -9,6 +9,7 @@
 #include <threads.h>
 #include <unistd.h>
 #include <ucontext.h>
+#include <inttypes.h>
 
 #include "arch/context.h"
 #include "current_sandbox.h"
@@ -25,6 +26,9 @@
 #include "software_interrupt.h"
 #include "software_interrupt_counts.h"
 
+extern uint64_t total_held[1024];
+extern uint64_t longest_held[1024];
+extern thread_local bool pthread_stop;
 thread_local _Atomic volatile sig_atomic_t handler_depth    = 0;
 thread_local _Atomic volatile sig_atomic_t deferred_sigalrm = 0;
 
@@ -33,7 +37,14 @@ thread_local _Atomic volatile sig_atomic_t deferred_sigalrm = 0;
  **************************************/
 
 extern pthread_t *runtime_worker_threads;
+extern time_t t_start;
+extern pthread_t listener_thread_id;
+extern thread_local uint32_t total_local_requests;
+extern thread_local int worker_thread_idx;
 
+#ifdef SANDBOX_STATE_TOTALS
+extern _Atomic uint32_t sandbox_state_totals[SANDBOX_STATE_COUNT];
+#endif
 /**************************
  * Private Static Inlines *
  *************************/
@@ -75,6 +86,32 @@ propagate_sigalrm(siginfo_t *signal_info)
 	}
 }
 
+/**
+ * A POSIX signal is delivered to only one thread.
+ * This function broadcasts the sigint signal to all other worker threads
+ */
+static inline void
+sigint_propagate_workers_listener(siginfo_t *signal_info)
+{
+        /* Signal was sent directly by the kernel user space, so forward to other threads */
+        if (signal_info->si_code == SI_KERNEL || signal_info->si_code == SI_USER) {
+                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);
+                        pthread_kill(runtime_worker_threads[i], SIGINT);
+                }
+                /* send to listener thread */
+                if (pthread_self() != listener_thread_id) {
+                        pthread_kill(listener_thread_id, SIGINT);
+                }
+        } else {
+                /* Signal forwarded from another thread. Just confirm it resulted from pthread_kill */
+                assert(signal_info->si_code == SI_TKILL);
+        }
+}
+
 static inline bool
 worker_thread_is_running_cooperative_scheduler(void)
 {
@@ -184,17 +221,31 @@ software_interrupt_handle_signals(int signal_type, siginfo_t *signal_info, void
 
 		break;
 	}
+	case SIGINT: {
+		/* Stop the alarm timer first */
+		software_interrupt_disarm_timer();
+		sigint_propagate_workers_listener(signal_info);
+   		/* calculate the throughput */
+		time_t t_end = time(NULL);
+		double seconds = difftime(t_end, t_start);	
+		//double throughput = atomic_load(&sandbox_state_totals[SANDBOX_COMPLETE]) / seconds;
+		double throughput = total_local_requests / seconds;
+	        uint32_t total_sandboxes_error = atomic_load(&sandbox_state_totals[SANDBOX_ERROR]); 	
+		printf("throughput is %f error request is %u global total request %d worker %d total requests is %u worker total_held %"PRIu64" longest_held %"PRIu64" listener total_held %"PRIu64" longest_held %"PRIu64"\n", throughput, total_sandboxes_error, atomic_load(&sandbox_state_totals[SANDBOX_COMPLETE]), worker_thread_idx, total_local_requests, total_held[worker_thread_idx], longest_held[worker_thread_idx], total_held[200], longest_held[200]);
+		fflush(stdout);
+		pthread_stop = true;		
+	}
 	default: {
 		const char *signal_name = strsignal(signal_type);
 		switch (signal_info->si_code) {
 		case SI_TKILL:
-			panic("software_interrupt_handle_signals unexpectedly received signal %s from a thread kill\n",
+			printf("software_interrupt_handle_signals unexpectedly received signal %s from a thread kill\n",
 			      signal_name);
 		case SI_KERNEL:
-			panic("software_interrupt_handle_signals unexpectedly received signal %s from the kernel\n",
+			printf("software_interrupt_handle_signals unexpectedly received signal %s from the kernel\n",
 			      signal_name);
 		default:
-			panic("software_interrupt_handle_signals unexpectedly received signal %s with si_code %d\n",
+			printf("software_interrupt_handle_signals unexpectedly received signal %s with si_code %d\n",
 			      signal_name, signal_info->si_code);
 		}
 	}
@@ -268,9 +319,10 @@ software_interrupt_initialize(void)
 	sigaddset(&signal_action.sa_mask, SIGUSR1);
 	sigaddset(&signal_action.sa_mask, SIGFPE);
 	sigaddset(&signal_action.sa_mask, SIGSEGV);
+	sigaddset(&signal_action.sa_mask, SIGINT);
 
-	const int    supported_signals[]   = { SIGALRM, SIGUSR1, SIGFPE, SIGSEGV };
-	const size_t supported_signals_len = 4;
+	const int    supported_signals[]   = { SIGALRM, SIGUSR1, SIGFPE, SIGSEGV, SIGINT };
+	const size_t supported_signals_len = 5;
 
 	for (int i = 0; i < supported_signals_len; i++) {
 		int signal = supported_signals[i];

runtime/src/worker_thread.c

@@ -18,6 +18,7 @@
 #include "tenant_functions.h"
 #include "priority_queue.h"
 
+extern thread_local int thread_id;
 /***************************
  * Worker Thread State     *
  **************************/
@@ -28,6 +29,8 @@ thread_local struct arch_context worker_thread_base_context;
 /* Used to index into global arguments and deadlines arrays */
 thread_local int worker_thread_idx;
 
+thread_local bool pthread_stop = false;
+
 /* Used to track tenants' timeouts */
 thread_local struct priority_queue *worker_thread_timeout_queue;
 /***********************
@@ -47,6 +50,7 @@ worker_thread_main(void *argument)
 
 	/* Index was passed via argument */
 	worker_thread_idx = *(int *)argument;
+	thread_id = worker_thread_idx;
 
 	/* Set my priority */
 	// runtime_set_pthread_prio(pthread_self(), 2);
@@ -65,6 +69,9 @@ worker_thread_main(void *argument)
 	software_interrupt_unmask_signal(SIGFPE);
 	software_interrupt_unmask_signal(SIGSEGV);
 
+	/* Unmask SIGINT signals */
+	software_interrupt_unmask_signal(SIGINT);
+
 	/* Unmask signals, unless the runtime has disabled preemption */
 	if (runtime_preemption_enabled) {
 		software_interrupt_unmask_signal(SIGALRM);