Merge pull request #325 from gwsystems/fix-scheduler-types

feat: Improve behavior when scheduling structures fill up

runtime/include/global_request_scheduler.h

@@ -5,7 +5,7 @@
 #include "sandbox_types.h"
 
 /* Returns pointer back if successful, null otherwise */
-typedef struct sandbox *(*global_request_scheduler_add_fn_t)(void *);
+typedef struct sandbox *(*global_request_scheduler_add_fn_t)(struct sandbox *);
 typedef int (*global_request_scheduler_remove_fn_t)(struct sandbox **);
 typedef int (*global_request_scheduler_remove_if_earlier_fn_t)(struct sandbox **, uint64_t);
 typedef uint64_t (*global_request_scheduler_peek_fn_t)(void);

runtime/include/priority_queue.h

@@ -64,8 +64,8 @@ priority_queue_append(struct priority_queue *priority_queue, void *new_item)
 
 	int rc;
 
-	if (unlikely(priority_queue->size + 1 > priority_queue->capacity)) panic("PQ overflow");
-	if (unlikely(priority_queue->size + 1 == priority_queue->capacity)) goto err_enospc;
+	if (unlikely(priority_queue->size > priority_queue->capacity)) panic("PQ overflow");
+	if (unlikely(priority_queue->size == priority_queue->capacity)) goto err_enospc;
 	priority_queue->items[++priority_queue->size] = new_item;
 
 	rc = 0;
@@ -271,16 +271,13 @@ priority_queue_initialize(size_t capacity, bool use_lock, priority_queue_get_pri
 	assert(get_priority_fn != NULL);
 
 	/* Add one to capacity because this data structure ignores the element at 0 */
-	size_t one_based_capacity = capacity + 1;
-
-	struct priority_queue *priority_queue = (struct priority_queue *)
-	  calloc(1, sizeof(struct priority_queue) + sizeof(void *) * one_based_capacity);
-
+	struct priority_queue *priority_queue = (struct priority_queue *)calloc(1, sizeof(struct priority_queue)
+	                                                                             + sizeof(void *) * (capacity + 1));
 
 	/* We're assuming a min-heap implementation, so set to larget possible value */
 	priority_queue_update_highest_priority(priority_queue, ULONG_MAX);
 	priority_queue->size            = 0;
-	priority_queue->capacity        = one_based_capacity; // Add one because we skip element 0
+	priority_queue->capacity        = capacity;
 	priority_queue->get_priority_fn = get_priority_fn;
 	priority_queue->use_lock        = use_lock;
 
@@ -289,6 +286,32 @@ priority_queue_initialize(size_t capacity, bool use_lock, priority_queue_get_pri
 	return priority_queue;
 }
 
+/**
+ * Double capacity of priority queue
+ * Note: currently there is no equivalent call for PQs that are not thread-local and need to be locked because it is
+ * unclear if the fact that the lock is a member in the struct that might be moved by realloc breaks the guarantees of
+ * the lock.
+ * @param priority_queue to resize
+ * @returns pointer to PR or NULL if realloc fails. This may have been moved by realloc!
+ */
+static inline struct priority_queue *
+priority_queue_grow_nolock(struct priority_queue *priority_queue)
+{
+	assert(priority_queue != NULL);
+
+	if (unlikely(priority_queue->capacity == 0)) {
+		priority_queue->capacity++;
+		debuglog("Growing to 1\n");
+	} else {
+		priority_queue->capacity *= 2;
+		debuglog("Growing to %zu\n", priority_queue->capacity);
+	}
+
+	/* capacity is padded by 1 because idx 0 is unused */
+	return (struct priority_queue *)realloc(priority_queue, sizeof(struct priority_queue)
+	                                                          + sizeof(void *) * (priority_queue->capacity + 1));
+}
+
 /**
  * Free the Priority Queue Data structure
  * @param priority_queue the priority_queue to initialize

runtime/src/global_request_scheduler.c

@@ -5,7 +5,7 @@
 
 /* Default uninitialized implementations of the polymorphic interface */
 noreturn static struct sandbox *
-uninitialized_add(void *arg)
+uninitialized_add(struct sandbox *arg)
 {
 	panic("Global Request Scheduler Add was called before initialization\n");
 }
@@ -43,6 +43,7 @@ global_request_scheduler_initialize(struct global_request_scheduler_config *conf
 /**
  * Adds a sandbox to the request scheduler
  * @param sandbox
+ * @returns pointer to sandbox if added. NULL otherwise
  */
 struct sandbox *
 global_request_scheduler_add(struct sandbox *sandbox)

runtime/src/global_request_scheduler_deque.c

@@ -12,7 +12,7 @@ static pthread_mutex_t global_request_scheduler_deque_mutex = PTHREAD_MUTEX_INIT
 /**
  * Pushes a sandbox to the global deque
  * @param sandbox_raw
- * @returns pointer to request if added. NULL otherwise
+ * @returns pointer to sandbox if added. NULL otherwise
  */
 static struct sandbox *
 global_request_scheduler_deque_add(struct sandbox *sandbox)

runtime/src/global_request_scheduler_minheap.c

@@ -12,19 +12,19 @@ static struct priority_queue *global_request_scheduler_minheap;
 /**
  * Pushes a sandbox to the global deque
  * @param sandbox
- * @returns pointer to request if added. Panics runtime otherwise
+ * @returns pointer to sandbox if added. NULL otherwise
  */
 static struct sandbox *
-global_request_scheduler_minheap_add(void *sandbox_raw)
+global_request_scheduler_minheap_add(struct sandbox *sandbox)
 {
-	assert(sandbox_raw);
+	assert(sandbox);
 	assert(global_request_scheduler_minheap);
 	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_raw);
-	/* TODO: Propagate -1 to caller. Issue #91 */
-	if (return_code == -ENOSPC) panic("Request Queue is full\n");
-	return (struct sandbox *)sandbox_raw;
+	int return_code = priority_queue_enqueue(global_request_scheduler_minheap, sandbox);
+
+	if (return_code != 0) return NULL;
+	return sandbox;
 }
 
 /**

runtime/src/listener_thread.c

@@ -189,8 +189,14 @@ listener_thread_main(void *dummy)
 					                    &sandbox->client_address);
 				}
 
-				/* Add to the Global Sandbox Request Scheduler */
-				global_request_scheduler_add(sandbox);
+				/* If the global request scheduler is full, return a 429 to the client */
+				sandbox = global_request_scheduler_add(sandbox);
+				if (unlikely(sandbox == NULL)) {
+					client_socket_send_oneshot(sandbox->client_socket_descriptor,
+					                           http_header_build(429), http_header_len(429));
+					client_socket_close(sandbox->client_socket_descriptor,
+					                    &sandbox->client_address);
+				}
 
 			} /* while true */
 		}         /* for loop */

runtime/src/local_runqueue_minheap.c

@@ -13,6 +13,8 @@
 #include "sandbox_functions.h"
 #include "runtime.h"
 
+#define INITIAL_LOCAL_RUNQUEUE_MINHEAP_CAPACITY 256
+
 thread_local static struct priority_queue *local_runqueue_minheap;
 
 /**
@@ -34,8 +36,13 @@ void
 local_runqueue_minheap_add(struct sandbox *sandbox)
 {
 	int return_code = priority_queue_enqueue_nolock(local_runqueue_minheap, sandbox);
-	/* TODO: propagate RC to caller. Issue #92 */
-	if (return_code == -ENOSPC) panic("Thread Runqueue is full!\n");
+	if (unlikely(return_code == -ENOSPC)) {
+		struct priority_queue *temp = priority_queue_grow_nolock(local_runqueue_minheap);
+		if (unlikely(temp == NULL)) panic("Failed to grow local runqueue\n");
+		local_runqueue_minheap = temp;
+		return_code            = priority_queue_enqueue_nolock(local_runqueue_minheap, sandbox);
+		if (unlikely(return_code == -ENOSPC)) panic("Thread Runqueue is full!\n");
+	}
 }
 
 /**
@@ -78,7 +85,8 @@ void
 local_runqueue_minheap_initialize()
 {
 	/* Initialize local state */
-	local_runqueue_minheap = priority_queue_initialize(256, false, sandbox_get_priority);
+	local_runqueue_minheap = priority_queue_initialize(INITIAL_LOCAL_RUNQUEUE_MINHEAP_CAPACITY, false,
+	                                                   sandbox_get_priority);
 
 	/* Register Function Pointers for Abstract Scheduling API */
 	struct local_runqueue_config config = { .add_fn      = local_runqueue_minheap_add,

tests/fibonacci/bimodal/Makefile

@@ -45,3 +45,6 @@ client-preempt:
 
 client-fib10-multi:
 	hey -z ${DURATION_SEC}s -cpus 4 -c 100 -t 0 -o csv -m GET -d "10\n" "http://${HOSTNAME}:10010"
+
+client-fib40-multi:
+	hey -z ${DURATION_SEC}s -cpus 4 -c 100 -t 0 -o csv -m GET -d "40\n" "http://${HOSTNAME}:10040"