Sean McBride
4 years ago
committed by
GitHub
commit
63ca27ab7f
@ -1,125 +1,14 @@
|
||||
#pragma once
|
||||
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdint.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "debuglog.h"
|
||||
#include "client_socket.h"
|
||||
|
||||
#define ADMISSIONS_CONTROL_GRANULARITY 1000000
|
||||
|
||||
/*
|
||||
* Unitless estimate of the instantaneous fraction of system capacity required to complete all previously
|
||||
* admitted work. This is used to calculate free capacity as part of admissions control
|
||||
*
|
||||
* The estimated requirements of a single admitted request is calculated as
|
||||
* estimated execution time (cycles) / relative deadline (cycles)
|
||||
*
|
||||
* These estimates are incremented on request acceptance and decremented on request completion (either
|
||||
* success or failure)
|
||||
*/
|
||||
extern _Atomic uint64_t admissions_control_admitted;
|
||||
extern uint64_t admissions_control_capacity;
|
||||
extern const double admissions_control_overhead;
|
||||
|
||||
static inline void
|
||||
admissions_control_initialize()
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
atomic_init(&admissions_control_admitted, 0);
|
||||
admissions_control_capacity = runtime_worker_threads_count * ADMISSIONS_CONTROL_GRANULARITY
|
||||
* ((double)1.0 - admissions_control_overhead);
|
||||
#endif
|
||||
}
|
||||
|
||||
static inline void
|
||||
admissions_control_add(uint64_t admissions_estimate)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(admissions_estimate > 0);
|
||||
atomic_fetch_add(&admissions_control_admitted, admissions_estimate);
|
||||
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Runtime Admitted: %lu / %lu\n", admissions_control_admitted, admissions_control_capacity);
|
||||
#endif
|
||||
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
static inline void
|
||||
admissions_control_subtract(uint64_t admissions_estimate)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
/* Assumption: Should never underflow */
|
||||
if (unlikely(admissions_estimate > admissions_control_admitted)) panic("Admissions Estimate underflow\n");
|
||||
|
||||
atomic_fetch_sub(&admissions_control_admitted, admissions_estimate);
|
||||
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Runtime Admitted: %lu / %lu\n", admissions_control_admitted, admissions_control_capacity);
|
||||
#endif
|
||||
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
|
||||
static inline uint64_t
|
||||
admissions_control_calculate_estimate(uint64_t estimated_execution, uint64_t relative_deadline)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(relative_deadline != 0);
|
||||
uint64_t admissions_estimate = (estimated_execution * (uint64_t)ADMISSIONS_CONTROL_GRANULARITY)
|
||||
/ relative_deadline;
|
||||
if (admissions_estimate == 0)
|
||||
panic("Ratio of Deadline to Execution time cannot exceed %d\n", ADMISSIONS_CONTROL_GRANULARITY);
|
||||
|
||||
return admissions_estimate;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
static inline uint64_t
|
||||
admissions_control_calculate_estimate_us(uint32_t estimated_execution_us, uint32_t relative_deadline_us)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(relative_deadline_us != 0);
|
||||
return (uint64_t)((uint64_t)(estimated_execution_us * ADMISSIONS_CONTROL_GRANULARITY)) / relative_deadline_us;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
static inline void
|
||||
admissions_control_log_decision(uint64_t admissions_estimate, bool admitted)
|
||||
{
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Admitted: %lu, Capacity: %lu, Estimate: %lu, Admitted? %s\n", admissions_control_admitted,
|
||||
admissions_control_capacity, admissions_estimate, admitted ? "yes" : "no");
|
||||
#endif /* LOG_ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
static inline uint64_t
|
||||
admissions_control_decide(uint64_t admissions_estimate)
|
||||
{
|
||||
uint64_t work_admitted = 1; /* Nominal non-zero value in case admissions control is disabled */
|
||||
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
if (unlikely(admissions_estimate == 0)) panic("Admissions estimate should never be zero");
|
||||
|
||||
uint64_t total_admitted = atomic_load(&admissions_control_admitted);
|
||||
|
||||
if (total_admitted + admissions_estimate >= admissions_control_capacity) {
|
||||
admissions_control_log_decision(admissions_estimate, false);
|
||||
work_admitted = 0;
|
||||
} else {
|
||||
admissions_control_log_decision(admissions_estimate, true);
|
||||
admissions_control_add(admissions_estimate);
|
||||
work_admitted = admissions_estimate;
|
||||
}
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
|
||||
return work_admitted;
|
||||
}
|
||||
void admissions_control_initialize();
|
||||
void admissions_control_add(uint64_t admissions_estimate);
|
||||
void admissions_control_subtract(uint64_t admissions_estimate);
|
||||
uint64_t admissions_control_calculate_estimate(uint64_t estimated_execution, uint64_t relative_deadline);
|
||||
uint64_t admissions_control_calculate_estimate_us(uint32_t estimated_execution_us, uint32_t relative_deadline_us);
|
||||
void admissions_control_log_decision(uint64_t admissions_estimate, bool admitted);
|
||||
uint64_t admissions_control_decide(uint64_t admissions_estimate);
|
||||
|
||||
@ -1,81 +1,7 @@
|
||||
#pragma once
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <errno.h>
|
||||
#include <unistd.h>
|
||||
#include <string.h>
|
||||
#include <sys/socket.h>
|
||||
|
||||
#include "panic.h"
|
||||
#include "debuglog.h"
|
||||
#include "http_response.h"
|
||||
#include "http_total.h"
|
||||
#include "runtime.h"
|
||||
#include "worker_thread.h"
|
||||
void client_socket_close(int client_socket, struct sockaddr *client_address);
|
||||
|
||||
|
||||
static inline void
|
||||
client_socket_close(int client_socket, struct sockaddr *client_address)
|
||||
{
|
||||
/* Should never close 0, 1, or 2 */
|
||||
assert(client_socket != STDIN_FILENO);
|
||||
assert(client_socket != STDOUT_FILENO);
|
||||
assert(client_socket != STDERR_FILENO);
|
||||
|
||||
|
||||
if (unlikely(close(client_socket) < 0)) {
|
||||
char client_address_text[INET6_ADDRSTRLEN] = {};
|
||||
if (unlikely(inet_ntop(AF_INET, &client_address, client_address_text, INET6_ADDRSTRLEN) == NULL)) {
|
||||
debuglog("Failed to log client_address: %s", strerror(errno));
|
||||
}
|
||||
debuglog("Error closing client socket %d associated with %s - %s", client_socket, client_address_text,
|
||||
strerror(errno));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Rejects request due to admission control or error
|
||||
* @param client_socket - the client we are rejecting
|
||||
* @param status_code - either 503 or 400
|
||||
*/
|
||||
static inline int
|
||||
client_socket_send(int client_socket, int status_code)
|
||||
{
|
||||
const char *response;
|
||||
int rc;
|
||||
switch (status_code) {
|
||||
case 503:
|
||||
response = HTTP_RESPONSE_503_SERVICE_UNAVAILABLE;
|
||||
http_total_increment_5XX();
|
||||
break;
|
||||
case 400:
|
||||
response = HTTP_RESPONSE_400_BAD_REQUEST;
|
||||
http_total_increment_4XX();
|
||||
break;
|
||||
default:
|
||||
panic("%d is not a valid status code\n", status_code);
|
||||
}
|
||||
|
||||
int sent = 0;
|
||||
int to_send = strlen(response);
|
||||
|
||||
while (sent < to_send) {
|
||||
rc = write(client_socket, &response[sent], to_send - sent);
|
||||
if (rc < 0) {
|
||||
if (errno == EAGAIN) { debuglog("Unexpectedly blocking on write of %s\n", response); }
|
||||
|
||||
debuglog("Error with %s\n", strerror(errno));
|
||||
|
||||
goto send_err;
|
||||
}
|
||||
sent += rc;
|
||||
};
|
||||
|
||||
rc = 0;
|
||||
done:
|
||||
return rc;
|
||||
send_err:
|
||||
debuglog("Error sending to client: %s", strerror(errno));
|
||||
rc = -1;
|
||||
goto done;
|
||||
}
|
||||
int client_socket_send(int client_socket, int status_code);
|
||||
|
||||
@ -1,10 +1,24 @@
|
||||
#pragma once
|
||||
|
||||
#include <stdbool.h>
|
||||
|
||||
#include "generic_thread.h"
|
||||
#include "module.h"
|
||||
|
||||
#define LISTENER_THREAD_CORE_ID 0
|
||||
|
||||
extern pthread_t listener_thread_id;
|
||||
|
||||
void listener_thread_initialize(void);
|
||||
__attribute__((noreturn)) void *listener_thread_main(void *dummy);
|
||||
int listener_thread_register_module(struct module *mod);
|
||||
|
||||
/**
|
||||
* Used to determine if running in the context of a listener thread
|
||||
* @returns true if listener. false if not (probably a worker)
|
||||
*/
|
||||
static inline bool
|
||||
listener_thread_is_running()
|
||||
{
|
||||
return pthread_self() == listener_thread_id;
|
||||
}
|
||||
|
||||
@ -1,6 +1,120 @@
|
||||
#include <stdatomic.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "admissions_control.h"
|
||||
#include "debuglog.h"
|
||||
#include "client_socket.h"
|
||||
|
||||
/*
|
||||
* Unitless estimate of the instantaneous fraction of system capacity required to complete all previously
|
||||
* admitted work. This is used to calculate free capacity as part of admissions control
|
||||
*
|
||||
* The estimated requirements of a single admitted request is calculated as
|
||||
* estimated execution time (cycles) / relative deadline (cycles)
|
||||
*
|
||||
* These estimates are incremented on request acceptance and decremented on request completion (either
|
||||
* success or failure)
|
||||
*/
|
||||
_Atomic uint64_t admissions_control_admitted;
|
||||
uint64_t admissions_control_capacity;
|
||||
|
||||
const double admissions_control_overhead = 0.2;
|
||||
|
||||
void
|
||||
admissions_control_initialize()
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
atomic_init(&admissions_control_admitted, 0);
|
||||
admissions_control_capacity = runtime_worker_threads_count * ADMISSIONS_CONTROL_GRANULARITY
|
||||
* ((double)1.0 - admissions_control_overhead);
|
||||
#endif
|
||||
}
|
||||
|
||||
void
|
||||
admissions_control_add(uint64_t admissions_estimate)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(admissions_estimate > 0);
|
||||
atomic_fetch_add(&admissions_control_admitted, admissions_estimate);
|
||||
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Runtime Admitted: %lu / %lu\n", admissions_control_admitted, admissions_control_capacity);
|
||||
#endif
|
||||
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
void
|
||||
admissions_control_subtract(uint64_t admissions_estimate)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
/* Assumption: Should never underflow */
|
||||
if (unlikely(admissions_estimate > admissions_control_admitted)) panic("Admissions Estimate underflow\n");
|
||||
|
||||
atomic_fetch_sub(&admissions_control_admitted, admissions_estimate);
|
||||
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Runtime Admitted: %lu / %lu\n", admissions_control_admitted, admissions_control_capacity);
|
||||
#endif
|
||||
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
uint64_t
|
||||
admissions_control_calculate_estimate(uint64_t estimated_execution, uint64_t relative_deadline)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(relative_deadline != 0);
|
||||
uint64_t admissions_estimate = (estimated_execution * (uint64_t)ADMISSIONS_CONTROL_GRANULARITY)
|
||||
/ relative_deadline;
|
||||
if (admissions_estimate == 0)
|
||||
panic("Ratio of Deadline to Execution time cannot exceed %d\n", ADMISSIONS_CONTROL_GRANULARITY);
|
||||
|
||||
return admissions_estimate;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
uint64_t
|
||||
admissions_control_calculate_estimate_us(uint32_t estimated_execution_us, uint32_t relative_deadline_us)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(relative_deadline_us != 0);
|
||||
return (uint64_t)((uint64_t)(estimated_execution_us * ADMISSIONS_CONTROL_GRANULARITY)) / relative_deadline_us;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
void
|
||||
admissions_control_log_decision(uint64_t admissions_estimate, bool admitted)
|
||||
{
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Admitted: %lu, Capacity: %lu, Estimate: %lu, Admitted? %s\n", admissions_control_admitted,
|
||||
admissions_control_capacity, admissions_estimate, admitted ? "yes" : "no");
|
||||
#endif /* LOG_ADMISSIONS_CONTROL */
|
||||
}
|
||||
|
||||
uint64_t
|
||||
admissions_control_decide(uint64_t admissions_estimate)
|
||||
{
|
||||
uint64_t work_admitted = 1; /* Nominal non-zero value in case admissions control is disabled */
|
||||
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
if (unlikely(admissions_estimate == 0)) panic("Admissions estimate should never be zero");
|
||||
|
||||
uint64_t total_admitted = atomic_load(&admissions_control_admitted);
|
||||
|
||||
if (total_admitted + admissions_estimate >= admissions_control_capacity) {
|
||||
admissions_control_log_decision(admissions_estimate, false);
|
||||
work_admitted = 0;
|
||||
} else {
|
||||
admissions_control_log_decision(admissions_estimate, true);
|
||||
admissions_control_add(admissions_estimate);
|
||||
work_admitted = admissions_estimate;
|
||||
}
|
||||
#endif /* ADMISSIONS_CONTROL */
|
||||
|
||||
return work_admitted;
|
||||
}
|
||||
|
||||
@ -0,0 +1,52 @@
|
||||
#include "admissions_info.h"
|
||||
#include "debuglog.h"
|
||||
|
||||
/**
|
||||
* Initializes perf window
|
||||
* @param self
|
||||
*/
|
||||
void
|
||||
admissions_info_initialize(struct admissions_info *self, int percentile, uint64_t expected_execution,
|
||||
uint64_t relative_deadline)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(relative_deadline > 0);
|
||||
assert(expected_execution > 0);
|
||||
self->relative_deadline = relative_deadline;
|
||||
self->estimate = admissions_control_calculate_estimate(expected_execution, relative_deadline);
|
||||
debuglog("Initial Estimate: %lu\n", self->estimate);
|
||||
assert(self != NULL);
|
||||
|
||||
perf_window_initialize(&self->perf_window);
|
||||
|
||||
if (unlikely(percentile < 50 || percentile > 99)) panic("Invalid admissions percentile");
|
||||
self->percentile = percentile;
|
||||
|
||||
self->control_index = PERF_WINDOW_BUFFER_SIZE * percentile / 100;
|
||||
#ifdef LOG_ADMISSIONS_CONTROL
|
||||
debuglog("Percentile: %d\n", self->percentile);
|
||||
debuglog("Control Index: %d\n", self->control_index);
|
||||
#endif
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
* Adds an execution value to the perf window and calculates and caches and updated estimate
|
||||
* @param self
|
||||
* @param execution_duration
|
||||
*/
|
||||
void
|
||||
admissions_info_update(struct admissions_info *self, uint64_t execution_duration)
|
||||
{
|
||||
#ifdef ADMISSIONS_CONTROL
|
||||
assert(!software_interrupt_is_enabled());
|
||||
struct perf_window *perf_window = &self->perf_window;
|
||||
|
||||
LOCK_LOCK(&self->perf_window.lock);
|
||||
perf_window_add(perf_window, execution_duration);
|
||||
uint64_t estimated_execution = perf_window_get_percentile(perf_window, self->percentile, self->control_index);
|
||||
self->estimate = admissions_control_calculate_estimate(estimated_execution, self->relative_deadline);
|
||||
LOCK_UNLOCK(&self->perf_window.lock);
|
||||
#endif
|
||||
}
|
||||
@ -0,0 +1,78 @@
|
||||
#include <arpa/inet.h>
|
||||
#include <errno.h>
|
||||
#include <string.h>
|
||||
#include <sys/socket.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "client_socket.h"
|
||||
#include "debuglog.h"
|
||||
#include "http_response.h"
|
||||
#include "http_total.h"
|
||||
#include "likely.h"
|
||||
#include "panic.h"
|
||||
#include "worker_thread.h"
|
||||
|
||||
void
|
||||
client_socket_close(int client_socket, struct sockaddr *client_address)
|
||||
{
|
||||
/* Should never close 0, 1, or 2 */
|
||||
assert(client_socket != STDIN_FILENO);
|
||||
assert(client_socket != STDOUT_FILENO);
|
||||
assert(client_socket != STDERR_FILENO);
|
||||
|
||||
if (unlikely(close(client_socket) < 0)) {
|
||||
char client_address_text[INET6_ADDRSTRLEN] = {};
|
||||
if (unlikely(inet_ntop(AF_INET, &client_address, client_address_text, INET6_ADDRSTRLEN) == NULL)) {
|
||||
debuglog("Failed to log client_address: %s", strerror(errno));
|
||||
}
|
||||
debuglog("Error closing client socket %d associated with %s - %s", client_socket, client_address_text,
|
||||
strerror(errno));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Rejects request due to admission control or error
|
||||
* @param client_socket - the client we are rejecting
|
||||
* @param status_code - either 503 or 400
|
||||
*/
|
||||
int
|
||||
client_socket_send(int client_socket, int status_code)
|
||||
{
|
||||
const char *response;
|
||||
int rc;
|
||||
switch (status_code) {
|
||||
case 503:
|
||||
response = HTTP_RESPONSE_503_SERVICE_UNAVAILABLE;
|
||||
http_total_increment_5XX();
|
||||
break;
|
||||
case 400:
|
||||
response = HTTP_RESPONSE_400_BAD_REQUEST;
|
||||
http_total_increment_4XX();
|
||||
break;
|
||||
default:
|
||||
panic("%d is not a valid status code\n", status_code);
|
||||
}
|
||||
|
||||
int sent = 0;
|
||||
int to_send = strlen(response);
|
||||
|
||||
while (sent < to_send) {
|
||||
rc = write(client_socket, &response[sent], to_send - sent);
|
||||
if (rc < 0) {
|
||||
if (errno == EAGAIN) { debuglog("Unexpectedly blocking on write of %s\n", response); }
|
||||
|
||||
debuglog("Error with %s\n", strerror(errno));
|
||||
|
||||
goto send_err;
|
||||
}
|
||||
sent += rc;
|
||||
};
|
||||
|
||||
rc = 0;
|
||||
done:
|
||||
return rc;
|
||||
send_err:
|
||||
debuglog("Error sending to client: %s", strerror(errno));
|
||||
rc = -1;
|
||||
goto done;
|
||||
}
|
||||
Loading…
Reference in new issue