diff --git a/example/blocking.c b/example/blocking.c index a145ed5..557bc87 100644 --- a/example/blocking.c +++ b/example/blocking.c @@ -27,12 +27,12 @@ int main() { ctx_reset, ctx_free); - lnm_http_step_append(&step, slow_step, true); + lnm_http_step_init(&step, slow_step); lnm_http_route_init_literal(&route, lnm_http_method_get, "/", step); lnm_http_loop_route_add(hl, route); lnm_log_init_global(); lnm_log_register_stdout(lnm_log_level_debug); - printf("res = %i\n", lnm_http_loop_run(hl, 8080, 1, 2)); + printf("res = %i\n", lnm_http_loop_run(hl, 8080, 1, 32)); } diff --git a/include/lnm/http/loop.h b/include/lnm/http/loop.h index fefdf6a..2427179 100644 --- a/include/lnm/http/loop.h +++ b/include/lnm/http/loop.h @@ -32,16 +32,23 @@ lnm_err lnm_http_loop_init(lnm_http_loop **out, void *c_gctx, lnm_http_ctx_reset_fn ctx_reset, lnm_http_ctx_free_fn ctx_free); +/** + * Initialize a new step. + * + * @param out where to store pointer to new `lnm_http_step` + * @param fn step function + */ +lnm_err lnm_http_step_init(lnm_http_step **out, lnm_http_step_fn fn); + /** * Append the given step fn to the step. * - * @param out both the previous step to append the new step to, and the output - * variable to which the new step is appended + * @param out where to store pointer to new `lnm_http_step` + * @param step step to append new step to * @param fn step function - * @param blocking whether the step is blocking or not */ -lnm_err lnm_http_step_append(lnm_http_step **out, lnm_http_step_fn fn, - bool blocking); +lnm_err lnm_http_step_append(lnm_http_step **out, lnm_http_step *step, + lnm_http_step_fn fn); /** * Initialize a new route of type literal. diff --git a/include/lnm/loop.h b/include/lnm/loop.h index b922c1f..781a745 100644 --- a/include/lnm/loop.h +++ b/include/lnm/loop.h @@ -12,11 +12,11 @@ #define LNM_QUEUE_MULTIPLIER 8 typedef enum lnm_loop_state { - lnm_loop_state_req_io = 0, - lnm_loop_state_res_io, + lnm_loop_state_req = 0, + lnm_loop_state_res, lnm_loop_state_end, - lnm_loop_state_req_work, - lnm_loop_state_res_work, + lnm_loop_state_req_blocking, + lnm_loop_state_res_blocking, } lnm_loop_state; /** @@ -106,26 +106,6 @@ lnm_err lnm_loop_run(lnm_loop *l); lnm_err lnm_loop_run_multi(lnm_loop *l, size_t epoll_threads, size_t worker_threads); -/** - * Advance the processing of the given connection. - * - * Behavior of this function depends on both the connection state and whether - * worker threads are enabled. - * - * For IO states, this function will perform network I/O along with executing - * the loop's respective processing functions. - * - * For work states, the respective processing functions are executed without - * performing any network I/O. If no worker queue is present, this function - * performs all blocking work until an I/O or the end state is reached. If there - * is a worker queue present, only one block of work is done before exiting, - * allowing further blocks of work to be scheduled on other worker threads. - * - * If no worker queue is present, this function will only exit once an I/O or - * end state is reached. - */ -void lnm_loop_conn_advance(lnm_loop *l, lnm_loop_conn *conn); - /** * Reschedule the given connection, either on the event loop for network IO or * on a worker thread for blocking work. Connections are terminated as needed. diff --git a/src/_include/lnm/loop_internal.h b/src/_include/lnm/loop_internal.h index c71f303..a5e70a8 100644 --- a/src/_include/lnm/loop_internal.h +++ b/src/_include/lnm/loop_internal.h @@ -9,6 +9,6 @@ void lnm_loop_conn_free(lnm_loop *l, lnm_loop_conn *conn); lnm_err lnm_loop_accept(lnm_loop *l); -void lnm_loop_conn_advance(lnm_loop *l, lnm_loop_conn *conn); +void lnm_loop_conn_io(lnm_loop *l, lnm_loop_conn *conn); #endif diff --git a/src/http/lnm_http_loop.c b/src/http/lnm_http_loop.c index bd83526..153b88c 100644 --- a/src/http/lnm_http_loop.c +++ b/src/http/lnm_http_loop.c @@ -28,8 +28,7 @@ lnm_err lnm_http_loop_init(lnm_http_loop **out, void *c_gctx, return lnm_err_ok; } -lnm_err lnm_http_step_append(lnm_http_step **out, lnm_http_step_fn fn, - bool blocking) { +lnm_err lnm_http_step_init(lnm_http_step **out, lnm_http_step_fn fn) { lnm_http_step *step = calloc(1, sizeof(lnm_http_step)); if (step == NULL) { @@ -37,17 +36,22 @@ lnm_err lnm_http_step_append(lnm_http_step **out, lnm_http_step_fn fn, } step->fn = fn; - step->blocking = blocking; - - if ((*out) != NULL) { - (*out)->next = step; - } - *out = step; return lnm_err_ok; } +lnm_err lnm_http_step_append(lnm_http_step **out, lnm_http_step *step, + lnm_http_step_fn fn) { + LNM_RES(lnm_http_step_init(out, fn)); + + if (step != NULL) { + step->next = *out; + } + + return lnm_err_ok; +} + lnm_err lnm_http_route_init(lnm_http_route **out) { lnm_http_route *route = calloc(1, sizeof(lnm_http_route)); diff --git a/src/http/lnm_http_loop_process.c b/src/http/lnm_http_loop_process.c index 1e465cf..40f7180 100644 --- a/src/http/lnm_http_loop_process.c +++ b/src/http/lnm_http_loop_process.c @@ -133,8 +133,8 @@ 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_work)) { - conn->state = lnm_loop_state_req_work; + if (step->blocking && (conn->state != lnm_loop_state_req_blocking)) { + conn->state = lnm_loop_state_req_blocking; break; } @@ -145,7 +145,7 @@ void lnm_http_loop_process_steps(lnm_http_conn *conn) { 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_io; + conn->state = lnm_loop_state_req; break; case lnm_http_step_err_close: conn->state = lnm_loop_state_end; @@ -157,7 +157,7 @@ void lnm_http_loop_process_steps(lnm_http_conn *conn) { } if (ctx->cur_step == NULL) { - conn->state = lnm_loop_state_res_io; + conn->state = lnm_loop_state_res; ctx->state = lnm_http_loop_state_add_headers; } } @@ -334,23 +334,23 @@ void (*process_fns[])(lnm_http_conn *conn) = { lnm_loop_state state_map[] = { // parse_req - lnm_loop_state_req_io, + lnm_loop_state_req, // route - lnm_loop_state_req_io, + lnm_loop_state_req, // parse_headers - lnm_loop_state_req_io, + lnm_loop_state_req, // steps - lnm_loop_state_req_io, + lnm_loop_state_req, // add_headers - lnm_loop_state_req_io, + lnm_loop_state_req, // write_status_line - lnm_loop_state_res_io, + lnm_loop_state_res, // write_headers - lnm_loop_state_res_io, + lnm_loop_state_res, // write_body - lnm_loop_state_res_io, + lnm_loop_state_res, // finish - lnm_loop_state_res_io, + lnm_loop_state_res, }; void lnm_http_loop_process(lnm_http_conn *conn) { @@ -378,7 +378,7 @@ void lnm_http_loop_process(lnm_http_conn *conn) { // We move the request to a dedicated buffer if the read buffer needs to be // reused - if ((conn->state == lnm_loop_state_req_io) && (conn->state == loop_state) && + if ((conn->state == lnm_loop_state_req) && (conn->state == loop_state) && (!ctx->req.buf.owned) && (ctx->req.buf.len > 0)) { char *buf = malloc(ctx->req.buf.len); diff --git a/src/loop/lnm_loop.c b/src/loop/lnm_loop.c index 5154b1e..ac6e5eb 100644 --- a/src/loop/lnm_loop.c +++ b/src/loop/lnm_loop.c @@ -55,7 +55,7 @@ lnm_err lnm_loop_accept(lnm_loop *l) { LNM_RES2(lnm_loop_conn_init(&conn, l), close(conn_fd)); conn->fd = conn_fd; - conn->state = lnm_loop_state_req_io; + conn->state = lnm_loop_state_req; struct epoll_event event = {.data.ptr = conn, .events = EPOLLIN | EPOLLET | EPOLLONESHOT}; @@ -132,17 +132,17 @@ lnm_err lnm_loop_setup(lnm_loop *l, uint16_t port) { void lnm_loop_conn_schedule(lnm_loop *l, lnm_loop_conn *conn) { switch (conn->state) { // IO states get rescheduled in the epoll loop - case lnm_loop_state_req_io: - case lnm_loop_state_res_io: { + case lnm_loop_state_req: + case lnm_loop_state_res: { struct epoll_event event = { .data.ptr = conn, - .events = (conn->state == lnm_loop_state_req_io ? EPOLLIN : EPOLLOUT) | + .events = (conn->state == lnm_loop_state_req ? EPOLLIN : EPOLLOUT) | EPOLLET | EPOLLONESHOT}; epoll_ctl(l->epoll_fd, EPOLL_CTL_MOD, conn->fd, &event); } break; - case lnm_loop_state_req_work: - case lnm_loop_state_res_work: + case lnm_loop_state_req_blocking: + case lnm_loop_state_res_blocking: lnm_loop_queue_push(l->wq, conn); break; case lnm_loop_state_end: { @@ -196,7 +196,7 @@ lnm_err lnm_loop_run(lnm_loop *l) { lnm_loop_conn *conn = events[i].data.ptr; // At this point, state is always an IO state - lnm_loop_conn_advance(l, conn); + lnm_loop_conn_io(l, conn); lnm_loop_conn_schedule(l, conn); } } diff --git a/src/loop/lnm_loop_io.c b/src/loop/lnm_loop_io.c index 58f5577..34da2d9 100644 --- a/src/loop/lnm_loop_io.c +++ b/src/loop/lnm_loop_io.c @@ -1,7 +1,7 @@ #include #include -#include #include +#include #include "lnm/loop.h" #include "lnm/loop_internal.h" @@ -34,7 +34,7 @@ void lnm_loop_conn_io_req(lnm_loop *l, lnm_loop_conn *conn) { conn->r.size += res; l->data_read(conn); - } while (conn->state == lnm_loop_state_req_io); + } while (conn->state == lnm_loop_state_req); } void lnm_loop_conn_io_res(lnm_loop *l, lnm_loop_conn *conn) { @@ -64,33 +64,17 @@ void lnm_loop_conn_io_res(lnm_loop *l, lnm_loop_conn *conn) { // writer function more space to work with memmove(conn->w.buf, &conn->w.buf[res], conn->w.size - res); conn->w.size -= res; - } while (conn->state == lnm_loop_state_res_io); + } while (conn->state == lnm_loop_state_res); } -void lnm_loop_conn_advance(lnm_loop *l, lnm_loop_conn *conn) { - do { - - switch (conn->state) { - case lnm_loop_state_req_io: - lnm_loop_conn_io_req(l, conn); - break; - case lnm_loop_state_res_io: - lnm_loop_conn_io_res(l, conn); - break; - case lnm_loop_state_req_work: - do { - l->data_read(conn); - } while (conn->state == lnm_loop_state_req_work); - break; - case lnm_loop_state_res_work: - do { - l->data_write(conn); - } while (conn->state == lnm_loop_state_res_work); - break; - default:; - } +void lnm_loop_conn_io(lnm_loop *l, lnm_loop_conn *conn) { + switch (conn->state) { + case lnm_loop_state_req: + lnm_loop_conn_io_req(l, conn); + break; + case lnm_loop_state_res: + lnm_loop_conn_io_res(l, conn); + break; + default:; } - // Execute all blocking work if we're running in single-threaded mode - while (l->wq == NULL && (conn->state == lnm_loop_state_req_work || - conn->state == lnm_loop_state_res_work)); } diff --git a/src/loop/lnm_loop_worker.c b/src/loop/lnm_loop_worker.c index fba4848..5e9d199 100644 --- a/src/loop/lnm_loop_worker.c +++ b/src/loop/lnm_loop_worker.c @@ -86,7 +86,16 @@ void lnm_loop_worker_run(void *arg) { lnm_loop_conn *conn = lnm_loop_queue_pop(q); lnm_ldebug("loop", "worker %i processing fd %i", thread_id, conn->fd); - lnm_loop_conn_advance(l, conn); + switch (conn->state) { + case lnm_loop_state_req_blocking: + l->data_read(conn); + break; + case lnm_loop_state_res_blocking: + l->data_write(conn); + // Other states shouldn't even end up here, so we ignore them + default:; + } + lnm_loop_conn_schedule(l, conn); } }