Sean McBride
4 years ago
parent
d8fa1fe4c7
commit
064dac1aaf
@ -1,7 +1,45 @@
|
||||
#pragma once
|
||||
|
||||
#include "sandbox.h"
|
||||
#include "sandbox_types.h"
|
||||
|
||||
struct sandbox *current_sandbox_get(void);
|
||||
void current_sandbox_set(struct sandbox *sandbox);
|
||||
void current_sandbox_block(void);
|
||||
/* current sandbox that is active.. */
|
||||
extern __thread struct sandbox *worker_thread_current_sandbox;
|
||||
|
||||
extern __thread struct sandbox_context_cache local_sandbox_context_cache;
|
||||
|
||||
void current_sandbox_start(void);
|
||||
|
||||
/**
|
||||
* Getter for the current sandbox executing on this thread
|
||||
* @returns the current sandbox executing on this thread
|
||||
*/
|
||||
static inline struct sandbox *
|
||||
current_sandbox_get(void)
|
||||
{
|
||||
return worker_thread_current_sandbox;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for the current sandbox executing on this thread
|
||||
* @param sandbox the sandbox we are setting this thread to run
|
||||
*/
|
||||
static inline void
|
||||
current_sandbox_set(struct sandbox *sandbox)
|
||||
{
|
||||
/* Unpack hierarchy to avoid pointer chasing */
|
||||
if (sandbox == NULL) {
|
||||
local_sandbox_context_cache = (struct sandbox_context_cache){
|
||||
.linear_memory_start = NULL,
|
||||
.linear_memory_size = 0,
|
||||
.module_indirect_table = NULL,
|
||||
};
|
||||
worker_thread_current_sandbox = NULL;
|
||||
} else {
|
||||
local_sandbox_context_cache = (struct sandbox_context_cache){
|
||||
.linear_memory_start = sandbox->linear_memory_start,
|
||||
.linear_memory_size = sandbox->linear_memory_size,
|
||||
.module_indirect_table = sandbox->module->indirect_table,
|
||||
};
|
||||
worker_thread_current_sandbox = sandbox;
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,37 @@
|
||||
#pragma once
|
||||
|
||||
#include <assert.h>
|
||||
#include <stddef.h>
|
||||
|
||||
#include "current_sandbox.h"
|
||||
#include "current_sandbox_yield.h"
|
||||
#include "generic_thread.h"
|
||||
#include "local_runqueue.h"
|
||||
#include "sandbox_functions.h"
|
||||
#include "software_interrupt.h"
|
||||
|
||||
/**
|
||||
* Mark the currently executing sandbox as blocked, remove it from the local runqueue,
|
||||
* and switch to base context
|
||||
*/
|
||||
static inline void
|
||||
current_sandbox_block(void)
|
||||
{
|
||||
software_interrupt_disable();
|
||||
|
||||
/* Remove the sandbox we were just executing from the runqueue and mark as blocked */
|
||||
struct sandbox *current_sandbox = current_sandbox_get();
|
||||
|
||||
assert(current_sandbox->state == SANDBOX_RUNNING);
|
||||
sandbox_set_as_blocked(current_sandbox, SANDBOX_RUNNING);
|
||||
generic_thread_dump_lock_overhead();
|
||||
|
||||
/* The worker thread seems to "spin" on a blocked sandbox, so try to execute another sandbox for one quantum
|
||||
* after blocking to give time for the action to resolve */
|
||||
struct sandbox *next_sandbox = local_runqueue_get_next();
|
||||
if (next_sandbox != NULL) {
|
||||
sandbox_switch_to(next_sandbox);
|
||||
} else {
|
||||
current_sandbox_yield();
|
||||
};
|
||||
}
|
||||
@ -0,0 +1,46 @@
|
||||
#pragma once
|
||||
|
||||
#include <assert.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
|
||||
#include "arch/context.h"
|
||||
#include "current_sandbox.h"
|
||||
#include "sandbox_types.h"
|
||||
#include "sandbox_functions.h"
|
||||
#include "software_interrupt.h"
|
||||
|
||||
/**
|
||||
* @brief Switches to the base context, placing the current sandbox on the completion queue if in RETURNED state
|
||||
*/
|
||||
static inline void
|
||||
current_sandbox_yield()
|
||||
{
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
struct sandbox *current_sandbox = current_sandbox_get();
|
||||
#ifndef NDEBUG
|
||||
if (current_sandbox != NULL) {
|
||||
assert(current_sandbox->state < SANDBOX_STATE_COUNT);
|
||||
assert(current_sandbox->stack_size == current_sandbox->module->stack_size);
|
||||
}
|
||||
#endif
|
||||
|
||||
/* Assumption: Base Context should never switch to Base Context */
|
||||
assert(current_sandbox != NULL);
|
||||
struct arch_context *current_context = ¤t_sandbox->ctxt;
|
||||
assert(current_context != &worker_thread_base_context);
|
||||
|
||||
#ifdef LOG_CONTEXT_SWITCHES
|
||||
debuglog("Sandbox %lu (@%p) (%s) > Base Context (@%p) (%s)\n", current_sandbox->id, current_context,
|
||||
arch_context_variant_print(current_sandbox->ctxt.variant), &worker_thread_base_context,
|
||||
arch_context_variant_print(worker_thread_base_context.variant));
|
||||
#endif
|
||||
|
||||
sandbox_exit(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();
|
||||
}
|
||||
@ -1,5 +1,3 @@
|
||||
#pragma once
|
||||
|
||||
#include "sandbox.h"
|
||||
|
||||
void local_runqueue_list_initialize();
|
||||
|
||||
@ -1,5 +1,3 @@
|
||||
#pragma once
|
||||
|
||||
#include "sandbox.h"
|
||||
|
||||
void local_runqueue_minheap_initialize();
|
||||
|
||||
@ -0,0 +1,576 @@
|
||||
#pragma once
|
||||
|
||||
#include <ucontext.h>
|
||||
#include <stdbool.h>
|
||||
|
||||
#include "arch/context.h"
|
||||
#include "client_socket.h"
|
||||
#include "current_sandbox.h"
|
||||
#include "deque.h"
|
||||
#include "http_parser.h"
|
||||
#include "http_request.h"
|
||||
#include "local_completion_queue.h"
|
||||
#include "local_runqueue.h"
|
||||
#include "module.h"
|
||||
#include "ps_list.h"
|
||||
#include "sandbox_request.h"
|
||||
#include "sandbox_state.h"
|
||||
#include "sandbox_types.h"
|
||||
#include "software_interrupt.h"
|
||||
|
||||
extern void current_sandbox_start(void);
|
||||
|
||||
/***************************
|
||||
* Public API *
|
||||
**************************/
|
||||
|
||||
struct sandbox *sandbox_allocate(struct sandbox_request *sandbox_request);
|
||||
void sandbox_close_http(struct sandbox *sandbox);
|
||||
void sandbox_free(struct sandbox *sandbox);
|
||||
void sandbox_free_linear_memory(struct sandbox *sandbox);
|
||||
int sandbox_initialize_file_descriptor(struct sandbox *sandbox);
|
||||
void sandbox_main(struct sandbox *sandbox);
|
||||
void sandbox_switch_to(struct sandbox *next_sandbox);
|
||||
|
||||
/**
|
||||
* Given a sandbox, returns the module that sandbox is executing
|
||||
* @param sandbox the sandbox whose module we want
|
||||
* @return the module of the provided sandbox
|
||||
*/
|
||||
static inline struct module *
|
||||
sandbox_get_module(struct sandbox *sandbox)
|
||||
{
|
||||
if (!sandbox) return NULL;
|
||||
return sandbox->module;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a sandbox's fd to the host fd it maps to
|
||||
* @param sandbox
|
||||
* @param fd_index index into the sandbox's fd table
|
||||
* @returns file descriptor or -1 in case of error
|
||||
*/
|
||||
static inline int
|
||||
sandbox_get_file_descriptor(struct sandbox *sandbox, int fd_index)
|
||||
{
|
||||
if (!sandbox) return -1;
|
||||
if (fd_index >= SANDBOX_MAX_FD_COUNT || fd_index < 0) return -1;
|
||||
return sandbox->file_descriptors[fd_index];
|
||||
}
|
||||
|
||||
static inline uint64_t
|
||||
sandbox_get_priority(void *element)
|
||||
{
|
||||
struct sandbox *sandbox = (struct sandbox *)element;
|
||||
return sandbox->absolute_deadline;
|
||||
};
|
||||
|
||||
/**
|
||||
* Maps a sandbox fd to an underlying host fd
|
||||
* Returns error condition if the file_descriptor to set does not contain sandbox preopen magic
|
||||
* @param sandbox
|
||||
* @param sandbox_fd index of the sandbox fd we want to set
|
||||
* @param file_descriptor the file descripter we want to set it to
|
||||
* @returns the index that was set or -1 in case of error
|
||||
*/
|
||||
static inline int
|
||||
sandbox_set_file_descriptor(struct sandbox *sandbox, int sandbox_fd, int host_fd)
|
||||
{
|
||||
if (!sandbox) return -1;
|
||||
if (sandbox_fd >= SANDBOX_MAX_FD_COUNT || sandbox_fd < 0) return -1;
|
||||
if (host_fd < 0 || sandbox->file_descriptors[sandbox_fd] != SANDBOX_FILE_DESCRIPTOR_PREOPEN_MAGIC) return -1;
|
||||
sandbox->file_descriptors[sandbox_fd] = host_fd;
|
||||
return sandbox_fd;
|
||||
}
|
||||
|
||||
/**
|
||||
* Map the host stdin, stdout, stderr to the sandbox
|
||||
* @param sandbox - the sandbox on which we are initializing stdio
|
||||
*/
|
||||
static inline void
|
||||
sandbox_initialize_stdio(struct sandbox *sandbox)
|
||||
{
|
||||
int sandbox_fd, rc;
|
||||
for (int host_fd = 0; host_fd <= 2; host_fd++) {
|
||||
sandbox_fd = sandbox_initialize_file_descriptor(sandbox);
|
||||
assert(sandbox_fd == host_fd);
|
||||
rc = sandbox_set_file_descriptor(sandbox, sandbox_fd, host_fd);
|
||||
assert(rc != -1);
|
||||
}
|
||||
}
|
||||
|
||||
static inline void
|
||||
sandbox_open_http(struct sandbox *sandbox)
|
||||
{
|
||||
assert(sandbox != NULL);
|
||||
|
||||
http_parser_init(&sandbox->http_parser, HTTP_REQUEST);
|
||||
|
||||
/* Set the sandbox as the data the http-parser has access to */
|
||||
sandbox->http_parser.data = sandbox;
|
||||
|
||||
/* Freshly allocated sandbox going runnable for first time, so register client socket with epoll */
|
||||
struct epoll_event accept_evt;
|
||||
accept_evt.data.ptr = (void *)sandbox;
|
||||
accept_evt.events = EPOLLIN | EPOLLOUT | EPOLLET;
|
||||
int rc = epoll_ctl(worker_thread_epoll_file_descriptor, EPOLL_CTL_ADD, sandbox->client_socket_descriptor,
|
||||
&accept_evt);
|
||||
if (unlikely(rc < 0)) panic_err();
|
||||
}
|
||||
|
||||
/**
|
||||
* Prints key performance metrics for a sandbox to runtime_sandbox_perf_log
|
||||
* This is defined by an environment variable
|
||||
* @param sandbox
|
||||
*/
|
||||
static inline void
|
||||
sandbox_print_perf(struct sandbox *sandbox)
|
||||
{
|
||||
/* If the log was not defined by an environment variable, early out */
|
||||
if (runtime_sandbox_perf_log == NULL) return;
|
||||
|
||||
uint32_t total_time_us = sandbox->total_time / runtime_processor_speed_MHz;
|
||||
uint32_t queued_us = (sandbox->allocation_timestamp - sandbox->request_arrival_timestamp)
|
||||
/ runtime_processor_speed_MHz;
|
||||
uint32_t initializing_us = sandbox->initializing_duration / runtime_processor_speed_MHz;
|
||||
uint32_t runnable_us = sandbox->runnable_duration / runtime_processor_speed_MHz;
|
||||
uint32_t running_us = sandbox->running_duration / runtime_processor_speed_MHz;
|
||||
uint32_t blocked_us = sandbox->blocked_duration / runtime_processor_speed_MHz;
|
||||
uint32_t returned_us = sandbox->returned_duration / runtime_processor_speed_MHz;
|
||||
|
||||
/*
|
||||
* Assumption: A sandbox is never able to free pages. If linear memory management
|
||||
* becomes more intelligent, then peak linear memory size needs to be tracked
|
||||
* seperately from current linear memory size.
|
||||
*/
|
||||
fprintf(runtime_sandbox_perf_log, "%lu,%s():%d,%s,%u,%u,%u,%u,%u,%u,%u,%u,%u\n", sandbox->id,
|
||||
sandbox->module->name, sandbox->module->port, sandbox_state_stringify(sandbox->state),
|
||||
sandbox->module->relative_deadline_us, total_time_us, queued_us, initializing_us, runnable_us,
|
||||
running_us, blocked_us, returned_us, sandbox->linear_memory_size);
|
||||
}
|
||||
|
||||
static inline void
|
||||
sandbox_summarize_page_allocations(struct sandbox *sandbox)
|
||||
{
|
||||
#ifdef LOG_SANDBOX_MEMORY_PROFILE
|
||||
// TODO: Handle interleavings
|
||||
char sandbox_page_allocations_log_path[100] = {};
|
||||
sandbox_page_allocations_log_path[99] = '\0';
|
||||
snprintf(sandbox_page_allocations_log_path, 99, "%s_%d_page_allocations.csv", sandbox->module->name,
|
||||
sandbox->module->port);
|
||||
|
||||
debuglog("Logging to %s", sandbox_page_allocations_log_path);
|
||||
|
||||
FILE *sandbox_page_allocations_log = fopen(sandbox_page_allocations_log_path, "a");
|
||||
|
||||
fprintf(sandbox_page_allocations_log, "%lu,%lu,%s,", sandbox->id, sandbox->running_duration,
|
||||
sandbox_state_stringify(sandbox->state));
|
||||
for (size_t i = 0; i < sandbox->page_allocation_timestamps_size; i++)
|
||||
fprintf(sandbox_page_allocations_log, "%u,", sandbox->page_allocation_timestamps[i]);
|
||||
|
||||
fprintf(sandbox_page_allocations_log, "\n");
|
||||
#else
|
||||
return;
|
||||
#endif
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_INITIALIZED state.
|
||||
* The sandbox was already zeroed out during allocation
|
||||
* @param sandbox an uninitialized sandbox
|
||||
* @param sandbox_request the request we are initializing the sandbox from
|
||||
* @param allocation_timestamp timestamp of allocation
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_initialized(struct sandbox *sandbox, struct sandbox_request *sandbox_request,
|
||||
uint64_t allocation_timestamp)
|
||||
{
|
||||
assert(!software_interrupt_is_enabled());
|
||||
assert(sandbox != NULL);
|
||||
assert(sandbox->state == SANDBOX_ALLOCATED);
|
||||
assert(sandbox_request != NULL);
|
||||
assert(allocation_timestamp > 0);
|
||||
|
||||
sandbox->id = sandbox_request->id;
|
||||
sandbox->admissions_estimate = sandbox_request->admissions_estimate;
|
||||
|
||||
sandbox->request_arrival_timestamp = sandbox_request->request_arrival_timestamp;
|
||||
sandbox->allocation_timestamp = allocation_timestamp;
|
||||
sandbox->state = SANDBOX_SET_AS_INITIALIZED;
|
||||
|
||||
/* Initialize the sandbox's context, stack, and instruction pointer */
|
||||
/* stack_start points to the bottom of the usable stack, so add stack_size to get to top */
|
||||
arch_context_init(&sandbox->ctxt, (reg_t)current_sandbox_start,
|
||||
(reg_t)sandbox->stack_start + sandbox->stack_size);
|
||||
|
||||
/* Mark sandbox fds as invalid by setting to -1 */
|
||||
for (int i = 0; i < SANDBOX_MAX_FD_COUNT; i++) sandbox->file_descriptors[i] = -1;
|
||||
|
||||
/* Initialize Parsec control structures */
|
||||
ps_list_init_d(sandbox);
|
||||
|
||||
/* Copy the socket descriptor, address, and arguments of the client invocation */
|
||||
sandbox->absolute_deadline = sandbox_request->absolute_deadline;
|
||||
sandbox->arguments = (void *)sandbox_request->arguments;
|
||||
sandbox->client_socket_descriptor = sandbox_request->socket_descriptor;
|
||||
memcpy(&sandbox->client_address, &sandbox_request->socket_address, sizeof(struct sockaddr));
|
||||
|
||||
sandbox->last_state_change_timestamp = allocation_timestamp; /* We use arg to include alloc */
|
||||
sandbox->state = SANDBOX_INITIALIZED;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, SANDBOX_UNINITIALIZED, SANDBOX_INITIALIZED);
|
||||
runtime_sandbox_total_increment(SANDBOX_INITIALIZED);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_RUNNABLE state.
|
||||
*
|
||||
* This occurs in the following scenarios:
|
||||
* - A sandbox in the SANDBOX_INITIALIZED state completes initialization and is ready to be run
|
||||
* - A sandbox in the SANDBOX_BLOCKED state completes what was blocking it and is ready to be run
|
||||
*
|
||||
* @param sandbox
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_runnable(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_RUNNABLE;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_INITIALIZED: {
|
||||
sandbox->initializing_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
case SANDBOX_BLOCKED: {
|
||||
sandbox->blocked_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Runnable\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
local_runqueue_add(sandbox);
|
||||
sandbox->last_state_change_timestamp = now;
|
||||
sandbox->state = SANDBOX_RUNNABLE;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, last_state, SANDBOX_RUNNABLE);
|
||||
runtime_sandbox_total_increment(SANDBOX_RUNNABLE);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
static inline void
|
||||
sandbox_set_as_running(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_RUNNING;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_RUNNABLE: {
|
||||
sandbox->runnable_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
case SANDBOX_PREEMPTED: {
|
||||
sandbox->preempted_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Running\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
current_sandbox_set(sandbox);
|
||||
sandbox->last_state_change_timestamp = now;
|
||||
sandbox->state = SANDBOX_RUNNING;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, last_state, SANDBOX_RUNNING);
|
||||
runtime_sandbox_total_increment(SANDBOX_RUNNING);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_BLOCKED state.
|
||||
* This occurs when a sandbox is executing and it makes a blocking API call of some kind.
|
||||
* Automatically removes the sandbox from the runqueue
|
||||
* @param sandbox the blocking sandbox
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_blocked(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_BLOCKED;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_RUNNING: {
|
||||
sandbox->running_duration += duration_of_last_state;
|
||||
local_runqueue_delete(sandbox);
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Blocked\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
sandbox->last_state_change_timestamp = now;
|
||||
sandbox->state = SANDBOX_BLOCKED;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, last_state, SANDBOX_BLOCKED);
|
||||
runtime_sandbox_total_increment(SANDBOX_BLOCKED);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_PREEMPTED state.
|
||||
*
|
||||
* This occurs when a sandbox is executing and in a RUNNING state and a SIGALRM software interrupt fires
|
||||
* and pulls a sandbox with an earlier absolute deadline from the global request scheduler.
|
||||
*
|
||||
* @param sandbox the sandbox being preempted
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_preempted(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_PREEMPTED;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_RUNNING: {
|
||||
sandbox->running_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Preempted\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
sandbox->last_state_change_timestamp = now;
|
||||
sandbox->state = SANDBOX_PREEMPTED;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, last_state, SANDBOX_PREEMPTED);
|
||||
runtime_sandbox_total_increment(SANDBOX_PREEMPTED);
|
||||
runtime_sandbox_total_decrement(SANDBOX_RUNNING);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_RETURNED state.
|
||||
* This occurs when a sandbox is executing and runs to completion.
|
||||
* Automatically removes the sandbox from the runqueue and unmaps linear memory.
|
||||
* Because the stack is still in use, freeing the stack is deferred until later
|
||||
* @param sandbox the blocking sandbox
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_returned(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_RETURNED;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_RUNNING: {
|
||||
sandbox->response_timestamp = now;
|
||||
sandbox->total_time = now - sandbox->request_arrival_timestamp;
|
||||
sandbox->running_duration += duration_of_last_state;
|
||||
local_runqueue_delete(sandbox);
|
||||
sandbox_free_linear_memory(sandbox);
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Returned\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
sandbox->last_state_change_timestamp = now;
|
||||
sandbox->state = SANDBOX_RETURNED;
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox->id, last_state, SANDBOX_RETURNED);
|
||||
runtime_sandbox_total_increment(SANDBOX_RETURNED);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox from the SANDBOX_RETURNED state to the SANDBOX_COMPLETE state.
|
||||
* Adds the sandbox to the completion queue
|
||||
* @param sandbox
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_complete(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_COMPLETE;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_RETURNED: {
|
||||
sandbox->completion_timestamp = now;
|
||||
sandbox->returned_duration += duration_of_last_state;
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Error\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t sandbox_id = sandbox->id;
|
||||
sandbox->state = SANDBOX_COMPLETE;
|
||||
sandbox_print_perf(sandbox);
|
||||
sandbox_summarize_page_allocations(sandbox);
|
||||
/* Admissions Control Post Processing */
|
||||
admissions_info_update(&sandbox->module->admissions_info, sandbox->running_duration);
|
||||
admissions_control_subtract(sandbox->admissions_estimate);
|
||||
/* Do not touch sandbox state after adding to completion queue to avoid use-after-free bugs */
|
||||
local_completion_queue_add(sandbox);
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox_id, last_state, SANDBOX_COMPLETE);
|
||||
runtime_sandbox_total_increment(SANDBOX_COMPLETE);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
/**
|
||||
* Transitions a sandbox to the SANDBOX_ERROR state.
|
||||
* This can occur during initialization or execution
|
||||
* Unmaps linear memory, removes from the runqueue (if on it), and adds to the completion queue
|
||||
* Because the stack is still in use, freeing the stack is deferred until later
|
||||
*
|
||||
* TODO: Is the sandbox adding itself to the completion queue here? Is this a problem? Issue #94
|
||||
*
|
||||
* @param sandbox the sandbox erroring out
|
||||
* @param last_state the state the sandbox is transitioning from. This is expressed as a constant to
|
||||
* enable the compiler to perform constant propagation optimizations.
|
||||
*/
|
||||
static inline void
|
||||
sandbox_set_as_error(struct sandbox *sandbox, sandbox_state_t last_state)
|
||||
{
|
||||
assert(sandbox);
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
uint64_t now = __getcycles();
|
||||
uint64_t duration_of_last_state = now - sandbox->last_state_change_timestamp;
|
||||
|
||||
sandbox->state = SANDBOX_SET_AS_ERROR;
|
||||
|
||||
switch (last_state) {
|
||||
case SANDBOX_SET_AS_INITIALIZED:
|
||||
/* Technically, this is a degenerate sandbox that we generate by hand */
|
||||
sandbox->initializing_duration += duration_of_last_state;
|
||||
break;
|
||||
case SANDBOX_RUNNING: {
|
||||
sandbox->running_duration += duration_of_last_state;
|
||||
local_runqueue_delete(sandbox);
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
panic("Sandbox %lu | Illegal transition from %s to Error\n", sandbox->id,
|
||||
sandbox_state_stringify(last_state));
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t sandbox_id = sandbox->id;
|
||||
sandbox->state = SANDBOX_ERROR;
|
||||
sandbox_print_perf(sandbox);
|
||||
sandbox_summarize_page_allocations(sandbox);
|
||||
sandbox_free_linear_memory(sandbox);
|
||||
admissions_control_subtract(sandbox->admissions_estimate);
|
||||
/* Do not touch sandbox after adding to completion queue to avoid use-after-free bugs */
|
||||
local_completion_queue_add(sandbox);
|
||||
|
||||
/* State Change Bookkeeping */
|
||||
sandbox_state_log_transition(sandbox_id, last_state, SANDBOX_ERROR);
|
||||
runtime_sandbox_total_increment(SANDBOX_ERROR);
|
||||
runtime_sandbox_total_decrement(last_state);
|
||||
}
|
||||
|
||||
/**
|
||||
* Conditionally triggers appropriate state changes for exiting sandboxes
|
||||
* @param exiting_sandbox - The sandbox that ran to completion
|
||||
*/
|
||||
static inline void
|
||||
sandbox_exit(struct sandbox *exiting_sandbox)
|
||||
{
|
||||
assert(exiting_sandbox != NULL);
|
||||
|
||||
switch (exiting_sandbox->state) {
|
||||
case SANDBOX_RETURNED:
|
||||
/*
|
||||
* We draw a distinction between RETURNED and COMPLETED because a sandbox cannot add itself to the
|
||||
* completion queue
|
||||
*/
|
||||
sandbox_set_as_complete(exiting_sandbox, SANDBOX_RETURNED);
|
||||
break;
|
||||
case SANDBOX_BLOCKED:
|
||||
/* Cooperative yield, so just break */
|
||||
break;
|
||||
case SANDBOX_ERROR:
|
||||
/* Terminal State, so just break */
|
||||
break;
|
||||
default:
|
||||
panic("Cooperatively switching from a sandbox in a non-terminal %s state\n",
|
||||
sandbox_state_stringify(exiting_sandbox->state));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a blocked sandbox as runnable and add it to the runqueue
|
||||
* @param sandbox the sandbox to check and update if blocked
|
||||
*/
|
||||
static inline void
|
||||
sandbox_wakeup(struct sandbox *sandbox)
|
||||
{
|
||||
assert(sandbox != NULL);
|
||||
assert(sandbox->state == SANDBOX_BLOCKED);
|
||||
|
||||
software_interrupt_disable();
|
||||
sandbox_set_as_runnable(sandbox, SANDBOX_BLOCKED);
|
||||
software_interrupt_enable();
|
||||
}
|
||||
@ -0,0 +1,100 @@
|
||||
#pragma once
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <assert.h>
|
||||
#include <errno.h>
|
||||
#include <stddef.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "current_sandbox.h"
|
||||
#include "current_sandbox_block.h"
|
||||
#include "debuglog.h"
|
||||
#include "http_parser.h"
|
||||
#include "http_request.h"
|
||||
#include "http_parser_settings.h"
|
||||
#include "likely.h"
|
||||
#include "sandbox_types.h"
|
||||
|
||||
/**
|
||||
* Receive and Parse the Request for the current sandbox
|
||||
* @return 0 if message parsing complete, -1 on error
|
||||
*/
|
||||
static inline int
|
||||
sandbox_receive_request(struct sandbox *sandbox)
|
||||
{
|
||||
assert(sandbox != NULL);
|
||||
assert(sandbox->module->max_request_size > 0);
|
||||
assert(sandbox->request_response_data_length == 0);
|
||||
|
||||
int rc = 0;
|
||||
|
||||
while (!sandbox->http_request.message_end) {
|
||||
/* Read from the Socket */
|
||||
|
||||
/* Structured to closely follow usage example at https://github.com/nodejs/http-parser */
|
||||
http_parser * parser = &sandbox->http_parser;
|
||||
const http_parser_settings *settings = http_parser_settings_get();
|
||||
|
||||
int fd = sandbox->client_socket_descriptor;
|
||||
char * buf = &sandbox->request_response_data[sandbox->request_response_data_length];
|
||||
size_t len = sandbox->module->max_request_size - sandbox->request_response_data_length;
|
||||
|
||||
ssize_t recved = recv(fd, buf, len, 0);
|
||||
|
||||
if (recved < 0) {
|
||||
if (errno == EAGAIN) {
|
||||
current_sandbox_block();
|
||||
continue;
|
||||
} else {
|
||||
/* All other errors */
|
||||
debuglog("Error reading socket %d - %s\n", sandbox->client_socket_descriptor,
|
||||
strerror(errno));
|
||||
goto err;
|
||||
}
|
||||
}
|
||||
|
||||
/* Client request is malformed */
|
||||
if (recved == 0 && !sandbox->http_request.message_end) {
|
||||
char client_address_text[INET6_ADDRSTRLEN] = {};
|
||||
if (unlikely(inet_ntop(AF_INET, &sandbox->client_address, client_address_text, INET6_ADDRSTRLEN)
|
||||
== NULL)) {
|
||||
debuglog("Failed to log client_address: %s", strerror(errno));
|
||||
}
|
||||
|
||||
debuglog("Sandbox %lu: recv returned 0 before a complete request was received\n", sandbox->id);
|
||||
debuglog("Socket: %d. Address: %s\n", fd, client_address_text);
|
||||
http_request_print(&sandbox->http_request);
|
||||
goto err;
|
||||
}
|
||||
|
||||
#ifdef LOG_HTTP_PARSER
|
||||
debuglog("Sandbox: %lu http_parser_execute(%p, %p, %p, %zu\n)", sandbox->id, parser, settings, buf,
|
||||
recved);
|
||||
#endif
|
||||
size_t nparsed = http_parser_execute(parser, settings, buf, recved);
|
||||
|
||||
if (nparsed != recved) {
|
||||
/* TODO: Is this error */
|
||||
debuglog("Error: %s, Description: %s\n",
|
||||
http_errno_name((enum http_errno)sandbox->http_parser.http_errno),
|
||||
http_errno_description((enum http_errno)sandbox->http_parser.http_errno));
|
||||
debuglog("Length Parsed %zu, Length Read %zu\n", nparsed, recved);
|
||||
debuglog("Error parsing socket %d\n", sandbox->client_socket_descriptor);
|
||||
goto err;
|
||||
}
|
||||
|
||||
|
||||
sandbox->request_response_data_length += nparsed;
|
||||
}
|
||||
|
||||
|
||||
sandbox->request_length = sandbox->request_response_data_length;
|
||||
|
||||
rc = 0;
|
||||
done:
|
||||
return rc;
|
||||
err:
|
||||
rc = -1;
|
||||
goto done;
|
||||
}
|
||||
@ -0,0 +1,121 @@
|
||||
#pragma once
|
||||
|
||||
#include <assert.h>
|
||||
#include <errno.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "current_sandbox.h"
|
||||
#include "current_sandbox_block.h"
|
||||
#include "http.h"
|
||||
#include "http_total.h"
|
||||
#include "likely.h"
|
||||
#include "sandbox_types.h"
|
||||
#include "panic.h"
|
||||
|
||||
/**
|
||||
* Sends Response Back to Client
|
||||
* @return RC. -1 on Failure
|
||||
*/
|
||||
static inline int
|
||||
sandbox_send_response(struct sandbox *sandbox)
|
||||
{
|
||||
assert(sandbox != NULL);
|
||||
|
||||
/*
|
||||
* At this point the HTTP Request has filled the buffer up to request_length, after which
|
||||
* the STDOUT of the sandbox has been appended. We assume that our HTTP Response header is
|
||||
* smaller than the HTTP Request header, which allows us to use memmove once without copying
|
||||
* to an intermediate buffer.
|
||||
*/
|
||||
memset(sandbox->request_response_data, 0, sandbox->request_length);
|
||||
|
||||
/*
|
||||
* We use this cursor to keep track of our position in the buffer and later assert that we
|
||||
* haven't overwritten body data.
|
||||
*/
|
||||
size_t response_cursor = 0;
|
||||
|
||||
/* Append 200 OK */
|
||||
strncpy(sandbox->request_response_data, HTTP_RESPONSE_200_OK, strlen(HTTP_RESPONSE_200_OK));
|
||||
response_cursor += strlen(HTTP_RESPONSE_200_OK);
|
||||
|
||||
/* Content Type */
|
||||
strncpy(sandbox->request_response_data + response_cursor, HTTP_RESPONSE_CONTENT_TYPE,
|
||||
strlen(HTTP_RESPONSE_CONTENT_TYPE));
|
||||
response_cursor += strlen(HTTP_RESPONSE_CONTENT_TYPE);
|
||||
|
||||
/* Custom content type if provided, text/plain by default */
|
||||
if (strlen(sandbox->module->response_content_type) <= 0) {
|
||||
strncpy(sandbox->request_response_data + response_cursor, HTTP_RESPONSE_CONTENT_TYPE_PLAIN,
|
||||
strlen(HTTP_RESPONSE_CONTENT_TYPE_PLAIN));
|
||||
response_cursor += strlen(HTTP_RESPONSE_CONTENT_TYPE_PLAIN);
|
||||
} else {
|
||||
strncpy(sandbox->request_response_data + response_cursor, sandbox->module->response_content_type,
|
||||
strlen(sandbox->module->response_content_type));
|
||||
response_cursor += strlen(sandbox->module->response_content_type);
|
||||
}
|
||||
|
||||
strncpy(sandbox->request_response_data + response_cursor, HTTP_RESPONSE_CONTENT_TYPE_TERMINATOR,
|
||||
strlen(HTTP_RESPONSE_CONTENT_TYPE_TERMINATOR));
|
||||
response_cursor += strlen(HTTP_RESPONSE_CONTENT_TYPE_TERMINATOR);
|
||||
|
||||
/* Content Length */
|
||||
strncpy(sandbox->request_response_data + response_cursor, HTTP_RESPONSE_CONTENT_LENGTH,
|
||||
strlen(HTTP_RESPONSE_CONTENT_LENGTH));
|
||||
response_cursor += strlen(HTTP_RESPONSE_CONTENT_LENGTH);
|
||||
|
||||
size_t body_size = sandbox->request_response_data_length - sandbox->request_length;
|
||||
|
||||
char len[10] = { 0 };
|
||||
sprintf(len, "%zu", body_size);
|
||||
strncpy(sandbox->request_response_data + response_cursor, len, strlen(len));
|
||||
response_cursor += strlen(len);
|
||||
|
||||
strncpy(sandbox->request_response_data + response_cursor, HTTP_RESPONSE_CONTENT_LENGTH_TERMINATOR,
|
||||
strlen(HTTP_RESPONSE_CONTENT_LENGTH_TERMINATOR));
|
||||
response_cursor += strlen(HTTP_RESPONSE_CONTENT_LENGTH_TERMINATOR);
|
||||
|
||||
/*
|
||||
* Assumption: Our response header is smaller than the request header, so we do not overwrite
|
||||
* actual data that the program appended to the HTTP Request. If proves to be a bad assumption,
|
||||
* we have to copy the STDOUT string to a temporary buffer before writing the header
|
||||
*/
|
||||
if (unlikely(response_cursor >= sandbox->request_length)) {
|
||||
panic("Response Cursor: %zd is less that Request Length: %zd\n", response_cursor,
|
||||
sandbox->request_length);
|
||||
}
|
||||
|
||||
/* Move the Sandbox's Data after the HTTP Response Data */
|
||||
memmove(sandbox->request_response_data + response_cursor,
|
||||
sandbox->request_response_data + sandbox->request_length, body_size);
|
||||
response_cursor += body_size;
|
||||
|
||||
/* Capture Timekeeping data for end-to-end latency */
|
||||
uint64_t end_time = __getcycles();
|
||||
sandbox->total_time = end_time - sandbox->request_arrival_timestamp;
|
||||
|
||||
int rc;
|
||||
int sent = 0;
|
||||
while (sent < response_cursor) {
|
||||
rc = write(sandbox->client_socket_descriptor, &sandbox->request_response_data[sent],
|
||||
response_cursor - sent);
|
||||
if (rc < 0) {
|
||||
if (errno == EAGAIN)
|
||||
current_sandbox_block();
|
||||
else {
|
||||
perror("write");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
sent += rc;
|
||||
}
|
||||
|
||||
http_total_increment_2xx();
|
||||
|
||||
return 0;
|
||||
}
|
||||
@ -0,0 +1,39 @@
|
||||
#pragma once
|
||||
|
||||
#include <assert.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "sandbox_types.h"
|
||||
|
||||
/**
|
||||
* Takes the arguments from the sandbox struct and writes them into the WebAssembly linear memory
|
||||
*/
|
||||
static inline void
|
||||
sandbox_setup_arguments(struct sandbox *sandbox)
|
||||
{
|
||||
assert(sandbox != NULL);
|
||||
int32_t argument_count = module_get_argument_count(sandbox->module);
|
||||
|
||||
/* whatever gregor has, to be able to pass arguments to a module! */
|
||||
sandbox->arguments_offset = local_sandbox_context_cache.linear_memory_size;
|
||||
assert(local_sandbox_context_cache.linear_memory_start == sandbox->linear_memory_start);
|
||||
expand_memory();
|
||||
|
||||
int32_t *array_ptr = (int32_t *)worker_thread_get_memory_ptr_void(sandbox->arguments_offset,
|
||||
argument_count * sizeof(int32_t));
|
||||
int32_t string_off = sandbox->arguments_offset + (argument_count * sizeof(int32_t));
|
||||
|
||||
for (int i = 0; i < argument_count; i++) {
|
||||
char * arg = (char *)sandbox->arguments + (i * MODULE_MAX_ARGUMENT_SIZE);
|
||||
size_t str_sz = strlen(arg) + 1;
|
||||
|
||||
array_ptr[i] = string_off;
|
||||
/* why get_memory_ptr_for_runtime?? */
|
||||
strncpy(get_memory_ptr_for_runtime(string_off, str_sz), arg, strlen(arg));
|
||||
|
||||
string_off += str_sz;
|
||||
}
|
||||
stub_init(string_off);
|
||||
}
|
||||
@ -1,451 +0,0 @@
|
||||
#include <assert.h>
|
||||
#include <errno.h>
|
||||
#include <limits.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
#include "listener_thread.h"
|
||||
#include "panic.h"
|
||||
#include "priority_queue.h"
|
||||
|
||||
/****************************
|
||||
* Private Helper Functions *
|
||||
***************************/
|
||||
|
||||
static inline void
|
||||
priority_queue_update_highest_priority(struct priority_queue *self, const uint64_t priority)
|
||||
{
|
||||
self->highest_priority = priority;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a value to the end of the binary heap
|
||||
* @param self the priority queue
|
||||
* @param new_item the value we are adding
|
||||
* @return 0 on success. -ENOSPC when priority queue is full
|
||||
*/
|
||||
static inline int
|
||||
priority_queue_append(struct priority_queue *self, void *new_item)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(new_item != NULL);
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
int rc;
|
||||
|
||||
if (unlikely(self->size + 1 > self->capacity)) panic("PQ overflow");
|
||||
if (unlikely(self->size + 1 == self->capacity)) goto err_enospc;
|
||||
self->items[++self->size] = new_item;
|
||||
|
||||
rc = 0;
|
||||
done:
|
||||
return rc;
|
||||
err_enospc:
|
||||
rc = -ENOSPC;
|
||||
goto done;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if a priority queue is empty
|
||||
* @param self the priority queue to check
|
||||
* @returns true if empty, else otherwise
|
||||
*/
|
||||
static inline bool
|
||||
priority_queue_is_empty(struct priority_queue *self)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
assert(listener_thread_is_running() || !software_interrupt_is_enabled());
|
||||
|
||||
return self->size == 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shifts an appended value upwards to restore heap structure property
|
||||
* @param self the priority queue
|
||||
*/
|
||||
static inline void
|
||||
priority_queue_percolate_up(struct priority_queue *self)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(self->get_priority_fn != NULL);
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
/* If there's only one element, set memoized lookup and early out */
|
||||
if (self->size == 1) {
|
||||
priority_queue_update_highest_priority(self, self->get_priority_fn(self->items[1]));
|
||||
return;
|
||||
}
|
||||
|
||||
for (int i = self->size;
|
||||
i / 2 != 0 && self->get_priority_fn(self->items[i]) < self->get_priority_fn(self->items[i / 2]); i /= 2) {
|
||||
assert(self->get_priority_fn(self->items[i]) != ULONG_MAX);
|
||||
void *temp = self->items[i / 2];
|
||||
self->items[i / 2] = self->items[i];
|
||||
self->items[i] = temp;
|
||||
/* If percolated to highest priority, update highest priority */
|
||||
if (i / 2 == 1) priority_queue_update_highest_priority(self, self->get_priority_fn(self->items[1]));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the index of a node's smallest child
|
||||
* @param self the priority queue
|
||||
* @param parent_index
|
||||
* @returns the index of the smallest child
|
||||
*/
|
||||
static inline int
|
||||
priority_queue_find_smallest_child(struct priority_queue *self, const int parent_index)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(parent_index >= 1 && parent_index <= self->size);
|
||||
assert(self->get_priority_fn != NULL);
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
int left_child_index = 2 * parent_index;
|
||||
int right_child_index = 2 * parent_index + 1;
|
||||
assert(self->items[left_child_index] != NULL);
|
||||
|
||||
int smallest_child_idx;
|
||||
|
||||
/* If we don't have a right child or the left child is smaller, return it */
|
||||
if (right_child_index > self->size) {
|
||||
smallest_child_idx = left_child_index;
|
||||
} else if (self->get_priority_fn(self->items[left_child_index])
|
||||
< self->get_priority_fn(self->items[right_child_index])) {
|
||||
smallest_child_idx = left_child_index;
|
||||
} else {
|
||||
/* Otherwise, return the right child */
|
||||
smallest_child_idx = right_child_index;
|
||||
}
|
||||
|
||||
return smallest_child_idx;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shifts the top of the heap downwards. Used after placing the last value at
|
||||
* the top
|
||||
* @param self the priority queue
|
||||
*/
|
||||
static inline void
|
||||
priority_queue_percolate_down(struct priority_queue *self, int parent_index)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(self->get_priority_fn != NULL);
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
assert(!listener_thread_is_running());
|
||||
assert(!software_interrupt_is_enabled());
|
||||
|
||||
bool update_highest_value = parent_index == 1;
|
||||
|
||||
int left_child_index = 2 * parent_index;
|
||||
while (left_child_index >= 2 && left_child_index <= self->size) {
|
||||
int smallest_child_index = priority_queue_find_smallest_child(self, parent_index);
|
||||
/* Once the parent is equal to or less than its smallest child, break; */
|
||||
if (self->get_priority_fn(self->items[parent_index])
|
||||
<= self->get_priority_fn(self->items[smallest_child_index]))
|
||||
break;
|
||||
/* Otherwise, swap and continue down the tree */
|
||||
void *temp = self->items[smallest_child_index];
|
||||
self->items[smallest_child_index] = self->items[parent_index];
|
||||
self->items[parent_index] = temp;
|
||||
|
||||
parent_index = smallest_child_index;
|
||||
left_child_index = 2 * parent_index;
|
||||
}
|
||||
|
||||
/* Update memoized value if we touched the head */
|
||||
if (update_highest_value) {
|
||||
if (!priority_queue_is_empty(self)) {
|
||||
priority_queue_update_highest_priority(self, self->get_priority_fn(self->items[1]));
|
||||
} else {
|
||||
priority_queue_update_highest_priority(self, ULONG_MAX);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*********************
|
||||
* Public API *
|
||||
********************/
|
||||
|
||||
/**
|
||||
* Initialized the Priority Queue Data structure
|
||||
* @param capacity the number of elements to store in the data structure
|
||||
* @param use_lock indicates that we want a concurrent data structure
|
||||
* @param get_priority_fn pointer to a function that returns the priority of an element
|
||||
* @return priority queue
|
||||
*/
|
||||
struct priority_queue *
|
||||
priority_queue_initialize(size_t capacity, bool use_lock, priority_queue_get_priority_fn_t get_priority_fn)
|
||||
{
|
||||
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 *self = calloc(sizeof(struct priority_queue) + sizeof(void *) * one_based_capacity, 1);
|
||||
|
||||
|
||||
/* We're assuming a min-heap implementation, so set to larget possible value */
|
||||
priority_queue_update_highest_priority(self, ULONG_MAX);
|
||||
self->size = 0;
|
||||
self->capacity = one_based_capacity; // Add one because we skip element 0
|
||||
self->get_priority_fn = get_priority_fn;
|
||||
self->use_lock = use_lock;
|
||||
|
||||
if (use_lock) LOCK_INIT(&self->lock);
|
||||
|
||||
return self;
|
||||
}
|
||||
|
||||
/**
|
||||
* Free the Priority Queue Data structure
|
||||
* @param self the priority_queue to initialize
|
||||
*/
|
||||
void
|
||||
priority_queue_free(struct priority_queue *self)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(listener_thread_is_running() || !software_interrupt_is_enabled());
|
||||
|
||||
free(self);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self the priority_queue
|
||||
* @returns the number of elements in the priority queue
|
||||
*/
|
||||
int
|
||||
priority_queue_length_nolock(struct priority_queue *self)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(!listener_thread_is_running());
|
||||
assert(!software_interrupt_is_enabled());
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
return self->size;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self the priority_queue
|
||||
* @returns the number of elements in the priority queue
|
||||
*/
|
||||
int
|
||||
priority_queue_length(struct priority_queue *self)
|
||||
{
|
||||
LOCK_LOCK(&self->lock);
|
||||
int size = priority_queue_length_nolock(self);
|
||||
LOCK_UNLOCK(&self->lock);
|
||||
return size;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param value - the value we want to add
|
||||
* @returns 0 on success. -ENOSPC on full.
|
||||
*/
|
||||
int
|
||||
priority_queue_enqueue_nolock(struct priority_queue *self, void *value)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(value != NULL);
|
||||
assert(listener_thread_is_running() || !software_interrupt_is_enabled());
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
int rc;
|
||||
|
||||
if (unlikely(priority_queue_append(self, value) == -ENOSPC)) goto err_enospc;
|
||||
|
||||
priority_queue_percolate_up(self);
|
||||
|
||||
rc = 0;
|
||||
done:
|
||||
return rc;
|
||||
err_enospc:
|
||||
rc = -ENOSPC;
|
||||
goto done;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param value - the value we want to add
|
||||
* @returns 0 on success. -ENOSPC on full.
|
||||
*/
|
||||
int
|
||||
priority_queue_enqueue(struct priority_queue *self, void *value)
|
||||
{
|
||||
int rc;
|
||||
|
||||
LOCK_LOCK(&self->lock);
|
||||
rc = priority_queue_enqueue_nolock(self, value);
|
||||
LOCK_UNLOCK(&self->lock);
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to delete from
|
||||
* @param value - the value we want to delete
|
||||
* @returns 0 on success. -1 on not found
|
||||
*/
|
||||
int
|
||||
priority_queue_delete_nolock(struct priority_queue *self, void *value)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(value != NULL);
|
||||
assert(!listener_thread_is_running());
|
||||
assert(!software_interrupt_is_enabled());
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
for (int i = 1; i <= self->size; i++) {
|
||||
if (self->items[i] == value) {
|
||||
self->items[i] = self->items[self->size];
|
||||
self->items[self->size--] = NULL;
|
||||
priority_queue_percolate_down(self, i);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
return -1;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to delete from
|
||||
* @param value - the value we want to delete
|
||||
* @returns 0 on success. -1 on not found
|
||||
*/
|
||||
int
|
||||
priority_queue_delete(struct priority_queue *self, void *value)
|
||||
{
|
||||
int rc;
|
||||
|
||||
LOCK_LOCK(&self->lock);
|
||||
rc = priority_queue_delete_nolock(self, value);
|
||||
LOCK_UNLOCK(&self->lock);
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the dequeued element
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty
|
||||
*/
|
||||
int
|
||||
priority_queue_dequeue(struct priority_queue *self, void **dequeued_element)
|
||||
{
|
||||
return priority_queue_dequeue_if_earlier(self, dequeued_element, UINT64_MAX);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the dequeued element
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty
|
||||
*/
|
||||
int
|
||||
priority_queue_dequeue_nolock(struct priority_queue *self, void **dequeued_element)
|
||||
{
|
||||
return priority_queue_dequeue_if_earlier_nolock(self, dequeued_element, UINT64_MAX);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the dequeued element
|
||||
* @param target_deadline the deadline that the request must be earlier than in order to dequeue
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty or if none meet target_deadline
|
||||
*/
|
||||
int
|
||||
priority_queue_dequeue_if_earlier_nolock(struct priority_queue *self, void **dequeued_element, uint64_t target_deadline)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(dequeued_element != NULL);
|
||||
assert(self->get_priority_fn != NULL);
|
||||
assert(!listener_thread_is_running());
|
||||
assert(!software_interrupt_is_enabled());
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
int return_code;
|
||||
|
||||
/* If the dequeue is not higher priority (earlier timestamp) than targed_deadline, return immediately */
|
||||
if (priority_queue_is_empty(self) || self->highest_priority >= target_deadline) goto err_enoent;
|
||||
|
||||
*dequeued_element = self->items[1];
|
||||
self->items[1] = self->items[self->size];
|
||||
self->items[self->size--] = NULL;
|
||||
|
||||
priority_queue_percolate_down(self, 1);
|
||||
return_code = 0;
|
||||
|
||||
done:
|
||||
return return_code;
|
||||
err_enoent:
|
||||
return_code = -ENOENT;
|
||||
goto done;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the dequeued element
|
||||
* @param target_deadline the deadline that the request must be earlier than in order to dequeue
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty or if none meet target_deadline
|
||||
*/
|
||||
int
|
||||
priority_queue_dequeue_if_earlier(struct priority_queue *self, void **dequeued_element, uint64_t target_deadline)
|
||||
{
|
||||
int return_code;
|
||||
|
||||
LOCK_LOCK(&self->lock);
|
||||
return_code = priority_queue_dequeue_if_earlier_nolock(self, dequeued_element, target_deadline);
|
||||
LOCK_UNLOCK(&self->lock);
|
||||
|
||||
return return_code;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the top of the priority queue without removing it
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the top element
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty
|
||||
*/
|
||||
int
|
||||
priority_queue_top_nolock(struct priority_queue *self, void **dequeued_element)
|
||||
{
|
||||
assert(self != NULL);
|
||||
assert(dequeued_element != NULL);
|
||||
assert(self->get_priority_fn != NULL);
|
||||
assert(!listener_thread_is_running());
|
||||
assert(!software_interrupt_is_enabled());
|
||||
assert(!self->use_lock || LOCK_IS_LOCKED(&self->lock));
|
||||
|
||||
int return_code;
|
||||
|
||||
if (priority_queue_is_empty(self)) goto err_enoent;
|
||||
|
||||
*dequeued_element = self->items[1];
|
||||
return_code = 0;
|
||||
|
||||
done:
|
||||
return return_code;
|
||||
err_enoent:
|
||||
return_code = -ENOENT;
|
||||
goto done;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the top of the priority queue without removing it
|
||||
* @param self - the priority queue we want to add to
|
||||
* @param dequeued_element a pointer to set to the top element
|
||||
* @returns RC 0 if successfully set dequeued_element, -ENOENT if empty
|
||||
*/
|
||||
int
|
||||
priority_queue_top(struct priority_queue *self, void **dequeued_element)
|
||||
{
|
||||
int return_code;
|
||||
|
||||
LOCK_LOCK(&self->lock);
|
||||
return_code = priority_queue_top_nolock(self, dequeued_element);
|
||||
LOCK_UNLOCK(&self->lock);
|
||||
|
||||
return return_code;
|
||||
}
|
||||
Loading…
Reference in new issue