feat: add blocking step support to http loop

blocking
Jef Roosens 2024-02-01 18:42:29 +01:00
parent 9c01548256
commit a96ae39523
Signed by: Jef Roosens
GPG Key ID: B75D4F293C7052DB
5 changed files with 36 additions and 8 deletions

View File

@ -34,5 +34,5 @@ int main() {
lnm_log_init_global();
lnm_log_register_stdout(lnm_log_level_debug);
printf("res = %i\n", lnm_http_loop_run(hl, 8080, 1, 1));
printf("res = %i\n", lnm_http_loop_run(hl, 8080, 1, 32));
}

View File

@ -47,6 +47,7 @@ typedef struct lnm_loop_queue {
} buf;
size_t head;
size_t tail;
bool empty;
pthread_mutex_t mutex;
pthread_cond_t cond;
} lnm_loop_queue;

View File

@ -8,6 +8,7 @@
typedef struct lnm_http_step {
lnm_http_step_fn fn;
struct lnm_http_step *next;
bool blocking;
} lnm_http_step;
typedef enum lnm_http_route_type {

View File

@ -133,11 +133,19 @@ void lnm_http_loop_process_steps(lnm_http_conn *conn) {
while ((ctx->cur_step != NULL) && (step != ctx->cur_step)) {
step = ctx->cur_step;
if (step->blocking && (conn->state != lnm_loop_state_req_blocking)) {
conn->state = lnm_loop_state_req_blocking;
break;
}
switch (step->fn(conn)) {
case lnm_http_step_err_done:
ctx->cur_step = ctx->cur_step->next;
break;
case lnm_http_step_err_io_needed:
// Ensure steps that require more I/O are executed on the event loop
conn->state = lnm_loop_state_req;
break;
case lnm_http_step_err_close:
conn->state = lnm_loop_state_end;
@ -149,6 +157,7 @@ void lnm_http_loop_process_steps(lnm_http_conn *conn) {
}
if (ctx->cur_step == NULL) {
conn->state = lnm_loop_state_res;
ctx->state = lnm_http_loop_state_add_headers;
}
}

View File

@ -1,9 +1,10 @@
#include <sys/epoll.h>
#include "lnm/log.h"
#include "lnm/loop.h"
lnm_err lnm_loop_queue_init(lnm_loop_queue **out, size_t cap) {
lnm_loop_conn **arr = malloc(cap * sizeof(lnm_loop_conn *));
lnm_loop_conn **arr = calloc(cap, sizeof(lnm_loop_conn *));
if (arr == NULL) {
return lnm_err_failed_alloc;
@ -20,6 +21,10 @@ lnm_err lnm_loop_queue_init(lnm_loop_queue **out, size_t cap) {
q->buf.arr = arr;
q->buf.len = cap;
q->tail = 0;
q->head = 0;
q->empty = true;
pthread_mutex_init(&q->mutex, NULL);
pthread_cond_init(&q->cond, NULL);
@ -31,14 +36,15 @@ lnm_err lnm_loop_queue_init(lnm_loop_queue **out, size_t cap) {
void lnm_loop_queue_push(lnm_loop_queue *q, lnm_loop_conn *conn) {
pthread_mutex_lock(&q->mutex);
while (q->tail == q->head) {
while (q->head == q->tail && !q->empty) {
pthread_cond_wait(&q->cond, &q->mutex);
}
q->buf.arr[q->tail] = conn;
q->buf.arr[q->head] = conn;
// Make sure the index wraps around
q->tail = (q->tail + 1) % q->buf.len;
q->head = (q->head + 1) % q->buf.len;
q->empty = false;
// Unlock mutex and signal to waiting threads
pthread_mutex_unlock(&q->mutex);
@ -48,12 +54,14 @@ void lnm_loop_queue_push(lnm_loop_queue *q, lnm_loop_conn *conn) {
lnm_loop_conn *lnm_loop_queue_pop(lnm_loop_queue *q) {
pthread_mutex_lock(&q->mutex);
while (q->tail == q->head) {
while (q->empty) {
pthread_cond_wait(&q->cond, &q->mutex);
}
lnm_loop_conn *out = q->buf.arr[q->head];
q->head = (q->head + 1) % q->buf.len;
lnm_loop_conn *out = q->buf.arr[q->tail];
q->tail = (q->tail + 1) % q->buf.len;
q->empty = q->tail == q->head;
// Unlock mutex and signal to waiting threads
pthread_mutex_unlock(&q->mutex);
@ -66,8 +74,17 @@ void lnm_loop_worker_run(void *arg) {
lnm_loop *l = arg;
lnm_loop_queue *q = l->wq;
// Get thread ID by incrementing counter
pthread_mutex_lock(&l->threads.mutex);
int thread_id = l->threads.worker_count;
l->threads.worker_count++;
pthread_mutex_unlock(&l->threads.mutex);
while (1) {
lnm_loop_conn *conn = lnm_loop_queue_pop(q);
lnm_ldebug("loop", "worker %i processing fd %i", thread_id, conn->fd);
switch (conn->state) {
case lnm_loop_state_req_blocking: