diff --git a/demos/bench.c b/demos/bench.c index ab90243..8b29848 100644 --- a/demos/bench.c +++ b/demos/bench.c @@ -1,5 +1,6 @@ #include "rpc/client.h" #include "rpc/server.h" +#include "rpc/trace.h" #include #include @@ -13,18 +14,36 @@ typedef struct bench_server { rpc_server *rpc; pthread_t thread; + uint32_t workers; } bench_server; +typedef struct bench_gate { + pthread_mutex_t mutex; + pthread_cond_t cond; + uint64_t ready; + uint64_t total; + int start; + int abort; +} bench_gate; + typedef struct worker_args { const char *host; char port[16]; uint64_t requests; uint64_t warmup; + uint64_t pipeline; uint64_t latency_offset; uint64_t *latencies_ns; + bench_gate *gate; + int started; int failed; } worker_args; +static int gate_ready_and_wait(bench_gate *gate); +static int gate_wait_ready(bench_gate *gate, uint64_t total); +static void gate_start(bench_gate *gate); +static void gate_abort(bench_gate *gate); + static uint64_t now_ns(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); @@ -48,8 +67,9 @@ static void *server_main(void *arg) { } static int start_server(bench_server *server, char port[16]) { - memset(server, 0, sizeof(*server)); if (rpc_server_init(&server->rpc) != 0 || + (server->workers != 0 && + rpc_server_set_workers(server->rpc, server->workers) != 0) || rpc_server_add_route(server->rpc, 1, add_handler, NULL) != 0 || rpc_server_bind(server->rpc, "127.0.0.1", "0") != 0 || rpc_server_listen(server->rpc) != 0) { @@ -87,11 +107,17 @@ static int run_one_call(rpc_client *client, rpc_writer *payload, int64_t expecte return rc; } +static int validate_response(rpc_value *values, size_t count, int64_t expected) { + return count == 1 && values[0].type == RPC_TYPE_I64 && values[0].as.i64 == expected; +} + static void *worker_main(void *arg) { worker_args *worker = arg; + int announced_gate = 0; rpc_client *client = NULL; if (rpc_client_connect(&client, worker->host, worker->port) != 0) { worker->failed = 1; + (void)gate_ready_and_wait(worker->gate); return NULL; } @@ -109,16 +135,92 @@ static void *worker_main(void *arg) { } } - for (uint64_t i = 0; i < worker->requests; ++i) { + announced_gate = 1; + if (gate_ready_and_wait(worker->gate) != 0) { + worker->failed = 1; + goto done; + } + + if (worker->pipeline <= 1) { + for (uint64_t i = 0; i < worker->requests; ++i) { + uint64_t start = now_ns(); + if (run_one_call(client, &payload, 42) != 0) { + worker->failed = 1; + goto done; + } + worker->latencies_ns[worker->latency_offset + i] = now_ns() - start; + } + goto done; + } + + uint64_t *call_ids = calloc(worker->pipeline, sizeof(*call_ids)); + uint64_t *starts = calloc(worker->pipeline, sizeof(*starts)); + if (!call_ids || !starts) { + free(call_ids); + free(starts); + worker->failed = 1; + goto done; + } + + uint64_t sent = 0; + uint64_t received = 0; + while (sent < worker->requests && sent - received < worker->pipeline) { uint64_t start = now_ns(); - if (run_one_call(client, &payload, 42) != 0) { + uint64_t call_id = 0; + if (rpc_client_send_call(client, 1, &payload, &call_id) != 0) { worker->failed = 1; - goto done; + goto pipeline_done; + } + uint64_t slot = call_id % worker->pipeline; + call_ids[slot] = call_id; + starts[slot] = start; + sent++; + } + + while (received < worker->requests) { + uint64_t call_id = 0; + rpc_value *values = NULL; + size_t count = 0; + if (rpc_client_recv_response(client, &call_id, &values, &count) != 0 || + !validate_response(values, count, 42)) { + rpc_values_free(values); + worker->failed = 1; + goto pipeline_done; + } + + uint64_t slot = call_id % worker->pipeline; + if (call_ids[slot] != call_id) { + rpc_values_free(values); + worker->failed = 1; + goto pipeline_done; + } + worker->latencies_ns[worker->latency_offset + received] = + now_ns() - starts[slot]; + rpc_values_free(values); + received++; + + if (sent < worker->requests) { + uint64_t start = now_ns(); + uint64_t next_call_id = 0; + if (rpc_client_send_call(client, 1, &payload, &next_call_id) != 0) { + worker->failed = 1; + goto pipeline_done; + } + slot = next_call_id % worker->pipeline; + call_ids[slot] = next_call_id; + starts[slot] = start; + sent++; } - worker->latencies_ns[worker->latency_offset + i] = now_ns() - start; } +pipeline_done: + free(call_ids); + free(starts); + done: + if (!announced_gate) { + (void)gate_ready_and_wait(worker->gate); + } rpc_writer_free(&payload); rpc_client_close(client); return NULL; @@ -130,7 +232,17 @@ static int cmp_u64(const void *a, const void *b) { return (x > y) - (x < y); } -static double ns_to_us(uint64_t ns) { return (double)ns / 1000.0; } +static void format_duration(uint64_t ns, char out[16]) { + if (ns < 1000ull) { + snprintf(out, 16, "%" PRIu64 " ns", ns); + } else if (ns < 1000000ull) { + snprintf(out, 16, "%.2f us", (double)ns / 1000.0); + } else if (ns < 1000000000ull) { + snprintf(out, 16, "%.2f ms", (double)ns / 1000000.0); + } else { + snprintf(out, 16, "%.3f s", (double)ns / 1000000000.0); + } +} static uint64_t percentile(const uint64_t *values, uint64_t count, double pct) { if (count == 0) { @@ -153,10 +265,67 @@ static uint64_t parse_u64_arg(const char *text, uint64_t fallback) { return (uint64_t)value; } +static void deadline_after_ms(struct timespec *deadline, long ms) { + clock_gettime(CLOCK_REALTIME, deadline); + deadline->tv_sec += ms / 1000; + deadline->tv_nsec += (ms % 1000) * 1000000l; + if (deadline->tv_nsec >= 1000000000l) { + deadline->tv_sec++; + deadline->tv_nsec -= 1000000000l; + } +} + +static int gate_ready_and_wait(bench_gate *gate) { + pthread_mutex_lock(&gate->mutex); + gate->ready++; + pthread_cond_broadcast(&gate->cond); + while (!gate->start && !gate->abort) { + pthread_cond_wait(&gate->cond, &gate->mutex); + } + int ok = !gate->abort; + pthread_mutex_unlock(&gate->mutex); + return ok ? 0 : -1; +} + +static int gate_wait_ready(bench_gate *gate, uint64_t total) { + pthread_mutex_lock(&gate->mutex); + gate->total = total; + while (gate->ready < gate->total && !gate->abort) { + struct timespec deadline; + deadline_after_ms(&deadline, 15000); + int rc = pthread_cond_timedwait(&gate->cond, &gate->mutex, &deadline); + if (rc == ETIMEDOUT) { + gate->abort = 1; + pthread_cond_broadcast(&gate->cond); + pthread_mutex_unlock(&gate->mutex); + return -1; + } + } + int ok = !gate->abort; + pthread_mutex_unlock(&gate->mutex); + return ok ? 0 : -1; +} + +static void gate_start(bench_gate *gate) { + pthread_mutex_lock(&gate->mutex); + gate->start = 1; + pthread_cond_broadcast(&gate->cond); + pthread_mutex_unlock(&gate->mutex); +} + +static void gate_abort(bench_gate *gate) { + pthread_mutex_lock(&gate->mutex); + gate->abort = 1; + pthread_cond_broadcast(&gate->cond); + pthread_mutex_unlock(&gate->mutex); +} + int main(int argc, char **argv) { uint64_t total_requests = argc > 1 ? parse_u64_arg(argv[1], 50000) : 50000; uint64_t clients = argc > 2 ? parse_u64_arg(argv[2], 1) : 1; uint64_t warmup_total = argc > 3 ? parse_u64_arg(argv[3], 1000) : 1000; + uint64_t server_workers = argc > 4 ? parse_u64_arg(argv[4], 0) : 0; + uint64_t pipeline = argc > 5 ? parse_u64_arg(argv[5], 1) : 1; if (clients > total_requests) { clients = total_requests; } @@ -173,6 +342,8 @@ int main(int argc, char **argv) { } bench_server server; + memset(&server, 0, sizeof(server)); + server.workers = (uint32_t)server_workers; char port[16]; if (start_server(&server, port) != 0) { fprintf(stderr, "failed to start benchmark server\n"); @@ -187,30 +358,63 @@ int main(int argc, char **argv) { uint64_t warmup_rem = warmup_total % clients; uint64_t request_base = total_requests / clients; uint64_t request_rem = total_requests % clients; + bench_gate gate = {0}; + pthread_mutex_init(&gate.mutex, NULL); + pthread_cond_init(&gate.cond, NULL); - uint64_t bench_start = now_ns(); + uint64_t started = 0; for (uint64_t i = 0; i < clients; ++i) { uint64_t count = request_base + (i < request_rem ? 1u : 0u); workers[i] = (worker_args){ .host = "127.0.0.1", .requests = count, .warmup = warmup_base + (i < warmup_rem ? 1u : 0u), + .pipeline = pipeline, .latency_offset = offset, .latencies_ns = latencies, + .gate = &gate, }; snprintf(workers[i].port, sizeof(workers[i].port), "%s", port); offset += count; if (pthread_create(&threads[i], NULL, worker_main, &workers[i]) != 0) { workers[i].failed = 1; + } else { + workers[i].started = 1; + started++; } } int failed = 0; + int bench_started = 0; + uint64_t bench_start = 0; + if (gate_wait_ready(&gate, started) != 0) { + fprintf(stderr, "benchmark workers did not become ready\n"); + failed = 1; + stop_server(&server); + } else { + for (uint64_t i = 0; i < clients; ++i) { + failed |= workers[i].failed; + } + if (failed) { + gate_abort(&gate); + stop_server(&server); + } else { + rpc_trace_reset(); + rpc_trace_set_enabled(1); + bench_start = now_ns(); + bench_started = 1; + gate_start(&gate); + } + } + for (uint64_t i = 0; i < clients; ++i) { - pthread_join(threads[i], NULL); + if (workers[i].started) { + pthread_join(threads[i], NULL); + } failed |= workers[i].failed; } - uint64_t bench_elapsed = now_ns() - bench_start; + rpc_trace_set_enabled(0); + uint64_t bench_elapsed = bench_started ? now_ns() - bench_start : 0; stop_server(&server); if (failed) { @@ -229,19 +433,36 @@ int main(int argc, char **argv) { double seconds = (double)bench_elapsed / 1000000000.0; double throughput = (double)total_requests / seconds; - double avg_us = ns_to_us(sum / total_requests); + char elapsed_buf[16]; + char avg_buf[16]; + char p50_buf[16]; + char p95_buf[16]; + char p99_buf[16]; + char max_buf[16]; + format_duration(bench_elapsed, elapsed_buf); + format_duration(sum / total_requests, avg_buf); + format_duration(percentile(latencies, total_requests, 50), p50_buf); + format_duration(percentile(latencies, total_requests, 95), p95_buf); + format_duration(percentile(latencies, total_requests, 99), p99_buf); + format_duration(latencies[total_requests - 1], max_buf); printf("requests: %" PRIu64 "\n", total_requests); printf("clients: %" PRIu64 "\n", clients); printf("warmup: %" PRIu64 "\n", warmup_total); - printf("elapsed: %.3f s\n", seconds); + printf("server workers: %" PRIu64 "%s\n", server_workers, + server_workers == 0 ? " (auto)" : ""); + printf("pipeline depth: %" PRIu64 "\n", pipeline); + printf("elapsed: %s\n", elapsed_buf); printf("throughput: %.0f req/s\n", throughput); - printf("latency avg: %.2f us\n", avg_us); - printf("latency p50: %.2f us\n", ns_to_us(percentile(latencies, total_requests, 50))); - printf("latency p95: %.2f us\n", ns_to_us(percentile(latencies, total_requests, 95))); - printf("latency p99: %.2f us\n", ns_to_us(percentile(latencies, total_requests, 99))); - printf("latency max: %.2f us\n", ns_to_us(latencies[total_requests - 1])); + printf("latency avg: %s\n", avg_buf); + printf("latency p50: %s\n", p50_buf); + printf("latency p95: %s\n", p95_buf); + printf("latency p99: %s\n", p99_buf); + printf("latency max: %s\n", max_buf); + rpc_trace_dump(stdout); + pthread_cond_destroy(&gate.cond); + pthread_mutex_destroy(&gate.mutex); free(latencies); free(threads); free(workers); diff --git a/demos/server.c b/demos/server.c index 66ad272..f6c3d66 100644 --- a/demos/server.c +++ b/demos/server.c @@ -37,8 +37,10 @@ static int echo_string(rpc_ctx *ctx, const rpc_value *args, size_t argc, int main(int argc, char **argv) { const char *host = argc > 1 ? argv[1] : "127.0.0.1"; const char *port = argc > 2 ? argv[2] : "7000"; + uint32_t workers = argc > 3 ? (uint32_t)strtoul(argv[3], NULL, 10) : 0; if (rpc_server_init(&g_server) != 0 || + (workers != 0 && rpc_server_set_workers(g_server, workers) != 0) || rpc_server_add_route(g_server, 1, add_i64, NULL) != 0 || rpc_server_add_route(g_server, 2, echo_string, NULL) != 0 || rpc_server_bind(g_server, host, port) != 0 || diff --git a/include/rpc/client.h b/include/rpc/client.h index 65066c3..c2204ab 100644 --- a/include/rpc/client.h +++ b/include/rpc/client.h @@ -17,6 +17,10 @@ int rpc_client_ping(rpc_client *client); int rpc_client_call(rpc_client *client, uint32_t proc_id, const rpc_writer *args, rpc_value **out_values, size_t *out_count); +int rpc_client_send_call(rpc_client *client, uint32_t proc_id, + const rpc_writer *args, uint64_t *out_call_id); +int rpc_client_recv_response(rpc_client *client, uint64_t *out_call_id, + rpc_value **out_values, size_t *out_count); const char *rpc_client_error(const rpc_client *client); diff --git a/include/rpc/server.h b/include/rpc/server.h index a1236cd..6f3420e 100644 --- a/include/rpc/server.h +++ b/include/rpc/server.h @@ -18,6 +18,7 @@ uint32_t rpc_ctx_proc_id(const rpc_ctx *ctx); void rpc_ctx_yield(rpc_ctx *ctx); int rpc_server_init(rpc_server **out_server); +int rpc_server_set_workers(rpc_server *server, uint32_t worker_count); int rpc_server_bind(rpc_server *server, const char *host, const char *port); uint16_t rpc_server_port(const rpc_server *server); int rpc_server_listen(rpc_server *server); diff --git a/include/rpc/trace.h b/include/rpc/trace.h new file mode 100644 index 0000000..99e3833 --- /dev/null +++ b/include/rpc/trace.h @@ -0,0 +1,87 @@ +#ifndef RPC_TRACE_H +#define RPC_TRACE_H + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef enum rpc_trace_metric { + RPC_TRACE_CLIENT_CALL, + RPC_TRACE_CLIENT_SEND, + RPC_TRACE_CLIENT_RECV, + RPC_TRACE_CLIENT_DECODE, + RPC_TRACE_SERVER_ACCEPT, + RPC_TRACE_SERVER_POLL_WAIT, + RPC_TRACE_SERVER_POLL_EVENTS, + RPC_TRACE_SERVER_LOOP_ACTIVE, + RPC_TRACE_SERVER_READ, + RPC_TRACE_SERVER_PARSE, + RPC_TRACE_SERVER_ROUTE, + RPC_TRACE_SERVER_SCHEDULE, + RPC_TRACE_SERVER_WRITE, + RPC_TRACE_SCHED_SUBMIT, + RPC_TRACE_SCHED_DECODE, + RPC_TRACE_SCHED_CORO_CREATE, + RPC_TRACE_SCHED_RESUME, + RPC_TRACE_PAYLOAD_DECODE, + RPC_TRACE_ROUTE_LOOKUP, + RPC_TRACE_COUNT, +} rpc_trace_metric; + +typedef struct rpc_trace_stat { + const char *name; + uint64_t count; + uint64_t total; + uint64_t max; + int is_time; +} rpc_trace_stat; + +extern int rpc_trace_enabled; + +void rpc_trace_set_enabled(int enabled); +uint64_t rpc_trace_begin_slow(void); +void rpc_trace_end_slow(rpc_trace_metric metric, uint64_t start_ns); +void rpc_trace_add_slow(rpc_trace_metric metric, uint64_t value); +void rpc_trace_reset(void); +void rpc_trace_snapshot(rpc_trace_stat out[RPC_TRACE_COUNT]); +void rpc_trace_dump(FILE *out); + +#if defined(__GNUC__) || defined(__clang__) +#define RPC_TRACE_UNLIKELY(x) __builtin_expect(!!(x), 0) +#else +#define RPC_TRACE_UNLIKELY(x) (x) +#endif + +static inline int rpc_trace_is_enabled(void) { +#if defined(__GNUC__) || defined(__clang__) + return RPC_TRACE_UNLIKELY( + __atomic_load_n(&rpc_trace_enabled, __ATOMIC_RELAXED)); +#else + return RPC_TRACE_UNLIKELY(rpc_trace_enabled); +#endif +} + +static inline uint64_t rpc_trace_begin(void) { + return rpc_trace_is_enabled() ? rpc_trace_begin_slow() : 0; +} + +static inline void rpc_trace_end(rpc_trace_metric metric, uint64_t start_ns) { + if (rpc_trace_is_enabled() && start_ns != 0) { + rpc_trace_end_slow(metric, start_ns); + } +} + +static inline void rpc_trace_add(rpc_trace_metric metric, uint64_t value) { + if (rpc_trace_is_enabled()) { + rpc_trace_add_slow(metric, value); + } +} + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/meson.build b/meson.build index 51681c7..f736885 100644 --- a/meson.build +++ b/meson.build @@ -18,6 +18,7 @@ core_sources = files( 'src/routes.c', 'src/scheduler.c', 'src/server.c', + 'src/trace.c', ) rpc_core = static_library('rpc_core', @@ -54,4 +55,4 @@ test_integration = executable('test_integration', 'tests/test_integration.c', test('protocol', test_protocol) test('routes', test_routes) test('scheduler', test_scheduler) -test('integration', test_integration, timeout: 10) \ No newline at end of file +test('integration', test_integration, timeout: 10) diff --git a/src/backend/kqueue.c b/src/backend/kqueue.c index 47df6be..a3dba38 100644 --- a/src/backend/kqueue.c +++ b/src/backend/kqueue.c @@ -7,6 +7,8 @@ #include #include +#define RPC_KQUEUE_MAX_EVENTS 1024 + struct rpc_backend { int kq; }; @@ -118,8 +120,8 @@ int rpc_backend_poll(rpc_backend *backend, rpc_backend_event *events, int max_ev timeout_ptr = &timeout; } - struct kevent kev[64]; - int limit = max_events < 64 ? max_events : 64; + struct kevent kev[RPC_KQUEUE_MAX_EVENTS]; + int limit = max_events < RPC_KQUEUE_MAX_EVENTS ? max_events : RPC_KQUEUE_MAX_EVENTS; int n = kevent(backend->kq, NULL, 0, kev, limit, timeout_ptr); if (n < 0) { if (errno == EINTR) { diff --git a/src/client.c b/src/client.c index 776795f..a834034 100644 --- a/src/client.c +++ b/src/client.c @@ -1,4 +1,5 @@ #include "rpc/client.h" +#include "rpc/trace.h" #include #include @@ -8,6 +9,12 @@ #include #include +#ifdef MSG_NOSIGNAL +#define RPC_SEND_FLAGS MSG_NOSIGNAL +#else +#define RPC_SEND_FLAGS 0 +#endif + struct rpc_client { int fd; uint64_t next_call_id; @@ -23,7 +30,7 @@ static void set_error(rpc_client *client, const char *message) { static int write_full(int fd, const void *data, size_t len) { const uint8_t *p = data; while (len > 0) { - ssize_t n = send(fd, p, len, 0); + ssize_t n = send(fd, p, len, RPC_SEND_FLAGS); if (n > 0) { p += n; len -= (size_t)n; @@ -56,6 +63,7 @@ static int read_full(int fd, void *data, size_t len) { static int send_packet(rpc_client *client, rpc_op op, uint32_t proc_id, uint64_t call_id, const rpc_writer *payload) { + uint64_t trace = rpc_trace_begin(); rpc_header header = { .op = op, .flags = RPC_FLAG_NONE, @@ -68,17 +76,21 @@ static int send_packet(rpc_client *client, rpc_op op, uint32_t proc_id, if (rpc_header_encode(&header, header_buf) != 0 || write_full(client->fd, header_buf, sizeof(header_buf)) != 0) { set_error(client, "send failed"); + rpc_trace_end(RPC_TRACE_CLIENT_SEND, trace); return -1; } if (payload && payload->len > 0 && write_full(client->fd, payload->data, payload->len) != 0) { set_error(client, "send failed"); + rpc_trace_end(RPC_TRACE_CLIENT_SEND, trace); return -1; } + rpc_trace_end(RPC_TRACE_CLIENT_SEND, trace); return 0; } static int recv_packet(rpc_client *client, rpc_header *header, uint8_t **body) { + uint64_t trace = rpc_trace_begin(); uint8_t header_buf[RPC_HEADER_SIZE]; *body = NULL; @@ -86,20 +98,24 @@ static int recv_packet(rpc_client *client, rpc_header *header, uint8_t **body) { rpc_header_decode(header_buf, header) != 0 || header->size > RPC_MAX_PAYLOAD_SIZE) { set_error(client, "read failed"); + rpc_trace_end(RPC_TRACE_CLIENT_RECV, trace); return -1; } *body = malloc(header->size ? header->size : 1); if (!*body) { set_error(client, "out of memory"); + rpc_trace_end(RPC_TRACE_CLIENT_RECV, trace); return -1; } if (read_full(client->fd, *body, header->size) != 0) { free(*body); *body = NULL; set_error(client, "read failed"); + rpc_trace_end(RPC_TRACE_CLIENT_RECV, trace); return -1; } + rpc_trace_end(RPC_TRACE_CLIENT_RECV, trace); return 0; } @@ -132,6 +148,11 @@ int rpc_client_connect(rpc_client **out_client, const char *host, if (fd < 0) { continue; } +#ifdef SO_NOSIGPIPE + int no_sigpipe = 1; + (void)setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &no_sigpipe, + sizeof(no_sigpipe)); +#endif if (connect(fd, ai->ai_addr, ai->ai_addrlen) == 0) { client->fd = fd; break; @@ -187,16 +208,55 @@ int rpc_client_ping(rpc_client *client) { int rpc_client_call(rpc_client *client, uint32_t proc_id, const rpc_writer *args, rpc_value **out_values, size_t *out_count) { + uint64_t trace_call = rpc_trace_begin(); if (!client || client->fd < 0 || !out_values || !out_count) { + rpc_trace_end(RPC_TRACE_CLIENT_CALL, trace_call); return -1; } *out_values = NULL; *out_count = 0; + uint64_t call_id = 0; + if (rpc_client_send_call(client, proc_id, args, &call_id) != 0) { + rpc_trace_end(RPC_TRACE_CLIENT_CALL, trace_call); + return -1; + } + + uint64_t response_call_id = 0; + int rc = rpc_client_recv_response(client, &response_call_id, out_values, out_count); + if (rc == 0 && response_call_id != call_id) { + rpc_values_free(*out_values); + *out_values = NULL; + *out_count = 0; + set_error(client, "unexpected call id"); + rc = -1; + } + + rpc_trace_end(RPC_TRACE_CLIENT_CALL, trace_call); + return rc; +} + +int rpc_client_send_call(rpc_client *client, uint32_t proc_id, + const rpc_writer *args, uint64_t *out_call_id) { + if (!client || client->fd < 0 || !out_call_id) { + return -1; + } uint64_t call_id = client->next_call_id++; if (send_packet(client, RPC_OP_RPC, proc_id, call_id, args) != 0) { return -1; } + *out_call_id = call_id; + return 0; +} + +int rpc_client_recv_response(rpc_client *client, uint64_t *out_call_id, + rpc_value **out_values, size_t *out_count) { + if (!client || client->fd < 0 || !out_call_id || !out_values || !out_count) { + return -1; + } + *out_call_id = 0; + *out_values = NULL; + *out_count = 0; rpc_header header; uint8_t *body = NULL; @@ -205,14 +265,15 @@ int rpc_client_call(rpc_client *client, uint32_t proc_id, } int rc = -1; - if (header.call_id != call_id) { - set_error(client, "unexpected call id"); - } else if (header.op == RPC_OP_RESPONSE) { + *out_call_id = header.call_id; + if (header.op == RPC_OP_RESPONSE) { + uint64_t trace_decode = rpc_trace_begin(); if (rpc_payload_decode(body, header.size, out_values, out_count) == 0) { rc = 0; } else { set_error(client, "malformed response payload"); } + rpc_trace_end(RPC_TRACE_CLIENT_DECODE, trace_decode); } else if (header.op == RPC_OP_ERROR) { rpc_value *values = NULL; size_t count = 0; diff --git a/src/payload.c b/src/payload.c index 27f07e2..d0e0fc2 100644 --- a/src/payload.c +++ b/src/payload.c @@ -1,4 +1,5 @@ #include "rpc/protocol.h" +#include "rpc/trace.h" #include #include @@ -148,6 +149,7 @@ int rpc_writer_string(rpc_writer *writer, const char *data, uint32_t len) { int rpc_payload_decode(const uint8_t *data, size_t len, rpc_value **out_values, size_t *out_count) { + uint64_t trace = rpc_trace_begin(); static const void *dispatch[] = { [RPC_TYPE_NULL] = &&type_null, [RPC_TYPE_BOOL] = &&type_bool, @@ -159,6 +161,7 @@ int rpc_payload_decode(const uint8_t *data, size_t len, rpc_value **out_values, }; if ((!data && len > 0) || !out_values || !out_count) { + rpc_trace_end(RPC_TRACE_PAYLOAD_DECODE, trace); return -1; } @@ -173,6 +176,7 @@ int rpc_payload_decode(const uint8_t *data, size_t len, rpc_value **out_values, rpc_value *next = realloc(values, next_cap * sizeof(*values)); if (!next) { free(values); + rpc_trace_end(RPC_TRACE_PAYLOAD_DECODE, trace); return -1; } values = next; @@ -252,11 +256,13 @@ store: malformed: free(values); + rpc_trace_end(RPC_TRACE_PAYLOAD_DECODE, trace); return -1; } *out_values = values; *out_count = count; + rpc_trace_end(RPC_TRACE_PAYLOAD_DECODE, trace); return 0; } diff --git a/src/routes.c b/src/routes.c index b55b7c2..53d6ee6 100644 --- a/src/routes.c +++ b/src/routes.c @@ -1,4 +1,5 @@ #include "routes.h" +#include "rpc/trace.h" #include #include @@ -150,7 +151,11 @@ int rpc_routes_remove(rpc_routes *routes, uint32_t proc_id) { } int rpc_routes_lookup(rpc_routes *routes, uint32_t proc_id, rpc_route *out) { - if (!routes || !out) return -1; + uint64_t trace = rpc_trace_begin(); + if (!routes || !out) { + rpc_trace_end(RPC_TRACE_ROUTE_LOOKUP, trace); + return -1; + } atomic_fetch_add_explicit(&routes->active_readers, 1u, memory_order_acquire); uint32_t page_idx = proc_id >> RPC_ROUTE_PAGE_BITS; @@ -169,5 +174,6 @@ int rpc_routes_lookup(rpc_routes *routes, uint32_t proc_id, rpc_route *out) { free_retired(routes); pthread_mutex_unlock(&routes->mutate_lock); } + rpc_trace_end(RPC_TRACE_ROUTE_LOOKUP, trace); return route ? 0 : -1; } diff --git a/src/scheduler.c b/src/scheduler.c index 568207e..2c5956e 100644 --- a/src/scheduler.c +++ b/src/scheduler.c @@ -1,6 +1,7 @@ #include "scheduler.h" #include "arena.h" +#include "rpc/trace.h" #define MCO_USE_VMEM_ALLOCATOR #define MCO_ZERO_MEMORY @@ -120,12 +121,15 @@ int rpc_scheduler_submit(rpc_scheduler *scheduler, uint64_t call_id, void *handler_data, const uint8_t *payload, size_t payload_len, rpc_call_done_fn done, void *done_data) { + uint64_t trace_submit = rpc_trace_begin(); if (!scheduler || !handler || (!payload && payload_len > 0)) { + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return -1; } rpc_call *call = rpc_fixed_arena_alloc(&scheduler->call_arena); if (!call) { + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return -1; } call->ctx.call_id = call_id; @@ -141,24 +145,31 @@ int rpc_scheduler_submit(rpc_scheduler *scheduler, uint64_t call_id, call->payload = malloc(payload_len); if (!call->payload) { call_free(scheduler, call); + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return -1; } memcpy(call->payload, payload, payload_len); } + uint64_t trace_decode = rpc_trace_begin(); if (rpc_payload_decode(call->payload, call->payload_len, &call->args, &call->argc) != 0) { + rpc_trace_end(RPC_TRACE_SCHED_DECODE, trace_decode); snprintf(call->error, sizeof(call->error), "malformed payload"); call->result = -1; call->completed = 1; done(call, done_data); call_free(scheduler, call); + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return 0; } + rpc_trace_end(RPC_TRACE_SCHED_DECODE, trace_decode); mco_desc desc = mco_desc_init(call_entry, 0); desc.user_data = call; + uint64_t trace_create = rpc_trace_begin(); mco_result rc = mco_create(&call->co, &desc); + rpc_trace_end(RPC_TRACE_SCHED_CORO_CREATE, trace_create); if (rc != MCO_SUCCESS) { snprintf(call->error, sizeof(call->error), "coroutine create failed: %s", mco_result_description(rc)); @@ -166,13 +177,16 @@ int rpc_scheduler_submit(rpc_scheduler *scheduler, uint64_t call_id, call->completed = 1; done(call, done_data); call_free(scheduler, call); + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return 0; } if (enqueue(scheduler, call) != 0) { call_free(scheduler, call); + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return -1; } + rpc_trace_end(RPC_TRACE_SCHED_SUBMIT, trace_submit); return 0; } @@ -183,7 +197,9 @@ void rpc_scheduler_run_ready(rpc_scheduler *scheduler) { rpc_call *call = NULL; while ((call = dequeue(scheduler)) != NULL) { + uint64_t trace_resume = rpc_trace_begin(); mco_result rc = mco_resume(call->co); + rpc_trace_end(RPC_TRACE_SCHED_RESUME, trace_resume); if (rc != MCO_SUCCESS) { snprintf(call->error, sizeof(call->error), "coroutine resume failed: %s", mco_result_description(rc)); diff --git a/src/server.c b/src/server.c index ac4fa40..2c9c6d6 100644 --- a/src/server.c +++ b/src/server.c @@ -1,4 +1,5 @@ #include "rpc/server.h" +#include "rpc/trace.h" #include "arena.h" #include "backend.h" @@ -10,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -19,17 +21,20 @@ #define RPC_MAX_LISTENERS 8 #define RPC_READ_CHUNK 4096 -#define RPC_EVENT_BATCH 64 +#define RPC_EVENT_BATCH 1024 #define RPC_CONNECTION_ARENA_CAPACITY 65536u +#define RPC_MAX_WORKERS 128u + +typedef struct rpc_worker rpc_worker; typedef struct rpc_listener { int fd; - rpc_server *server; + rpc_worker *worker; } rpc_listener; typedef struct rpc_connection { int fd; - rpc_server *server; + rpc_worker *worker; uint8_t *read_buf; size_t read_len; size_t read_cap; @@ -37,24 +42,49 @@ typedef struct rpc_connection { size_t write_len; size_t write_cap; size_t write_off; + uint32_t interests; int closing; struct rpc_connection *next; } rpc_connection; -struct rpc_server { +struct rpc_worker { + rpc_server *server; rpc_backend *backend; - rpc_routes routes; rpc_scheduler *scheduler; rpc_fixed_arena connection_arena; rpc_listener listeners[RPC_MAX_LISTENERS]; size_t listener_count; rpc_connection *connections; + pthread_t thread; + int thread_started; + uint32_t index; +}; + +struct rpc_server { + rpc_routes routes; + rpc_worker *workers; + uint32_t worker_count; + uint32_t workers_ready; atomic_bool stopping; int listening; int routes_ready; uint16_t port; }; +static void connection_write(rpc_connection *conn); + +static int connection_set_interests(rpc_connection *conn, uint32_t interests) { + if (conn->interests == interests) { + return 0; + } + if (rpc_backend_modify(conn->worker->backend, conn->fd, interests, + (uintptr_t)conn) != 0) { + return -1; + } + conn->interests = interests; + return 0; +} + static int set_nonblock(int fd) { int flags = fcntl(fd, F_GETFL, 0); if (flags < 0) { @@ -63,6 +93,35 @@ static int set_nonblock(int fd) { return fcntl(fd, F_SETFL, flags | O_NONBLOCK); } +static uint32_t cpu_count(void) { + long n = sysconf(_SC_NPROCESSORS_ONLN); + if (n <= 0) { + return 1; + } + if ((unsigned long)n > RPC_MAX_WORKERS) { + return RPC_MAX_WORKERS; + } + return (uint32_t)n; +} + +static void sockaddr_set_port(struct sockaddr *addr, uint16_t port) { + if (addr->sa_family == AF_INET) { + ((struct sockaddr_in *)addr)->sin_port = htons(port); + } else if (addr->sa_family == AF_INET6) { + ((struct sockaddr_in6 *)addr)->sin6_port = htons(port); + } +} + +static uint16_t sockaddr_get_port(const struct sockaddr *addr) { + if (addr->sa_family == AF_INET) { + return ntohs(((const struct sockaddr_in *)addr)->sin_port); + } + if (addr->sa_family == AF_INET6) { + return ntohs(((const struct sockaddr_in6 *)addr)->sin6_port); + } + return 0; +} + static int append_bytes(uint8_t **buf, size_t *len, size_t *cap, const void *data, size_t data_len) { if (data_len == 0) { @@ -133,9 +192,9 @@ static int queue_packet(rpc_connection *conn, rpc_op op, uint8_t flags, payload_len) != 0) { return -1; } - return rpc_backend_modify(conn->server->backend, conn->fd, - RPC_BACKEND_READ | RPC_BACKEND_WRITE, - (uintptr_t)conn); + + connection_write(conn); + return conn->closing ? -1 : 0; } static int queue_string_error(rpc_connection *conn, uint32_t proc_id, @@ -155,20 +214,20 @@ static void connection_destroy(rpc_connection *conn) { if (!conn) { return; } - rpc_server *server = conn->server; - (void)rpc_backend_remove(server->backend, conn->fd); + rpc_worker *worker = conn->worker; + (void)rpc_backend_remove(worker->backend, conn->fd); close(conn->fd); free(conn->read_buf); free(conn->write_buf); - rpc_connection **link = &server->connections; + rpc_connection **link = &worker->connections; while (*link && *link != conn) { link = &(*link)->next; } if (*link == conn) { *link = conn->next; } - rpc_fixed_arena_free(&server->connection_arena, conn); + rpc_fixed_arena_free(&worker->connection_arena, conn); } static void maybe_close(rpc_connection *conn) { @@ -216,17 +275,23 @@ op_disconnect: op_rpc: { rpc_route route; - if (rpc_routes_lookup(&conn->server->routes, header->proc_id, &route) != 0) { + uint64_t trace_route = rpc_trace_begin(); + if (rpc_routes_lookup(&conn->worker->server->routes, header->proc_id, &route) != 0) { + rpc_trace_end(RPC_TRACE_SERVER_ROUTE, trace_route); return queue_string_error(conn, header->proc_id, header->call_id, "unknown procedure"); } - if (rpc_scheduler_submit(conn->server->scheduler, header->call_id, + rpc_trace_end(RPC_TRACE_SERVER_ROUTE, trace_route); + uint64_t trace_schedule = rpc_trace_begin(); + if (rpc_scheduler_submit(conn->worker->scheduler, header->call_id, header->proc_id, route.handler, route.user_data, payload, header->size, on_call_done, conn) != 0) { + rpc_trace_end(RPC_TRACE_SERVER_SCHEDULE, trace_schedule); return queue_string_error(conn, header->proc_id, header->call_id, "scheduler failure"); } - rpc_scheduler_run_ready(conn->server->scheduler); + rpc_trace_end(RPC_TRACE_SERVER_SCHEDULE, trace_schedule); + rpc_scheduler_run_ready(conn->worker->scheduler); return 0; } @@ -236,6 +301,7 @@ op_unsupported: } static void parse_available(rpc_connection *conn) { + uint64_t trace = rpc_trace_begin(); size_t off = 0; while (conn->read_len - off >= RPC_HEADER_SIZE) { rpc_header header; @@ -258,14 +324,17 @@ static void parse_available(rpc_connection *conn) { memmove(conn->read_buf, conn->read_buf + off, conn->read_len - off); conn->read_len -= off; } + rpc_trace_end(RPC_TRACE_SERVER_PARSE, trace); } static void connection_read(rpc_connection *conn) { + uint64_t trace = rpc_trace_begin(); for (;;) { if (conn->read_cap - conn->read_len < RPC_READ_CHUNK) { if (reserve_bytes(&conn->read_buf, &conn->read_cap, conn->read_len + RPC_READ_CHUNK) != 0) { conn->closing = 1; + rpc_trace_end(RPC_TRACE_SERVER_READ, trace); return; } } @@ -275,26 +344,31 @@ static void connection_read(rpc_connection *conn) { conn->read_len += (size_t)n; parse_available(conn); if (conn->closing) { + rpc_trace_end(RPC_TRACE_SERVER_READ, trace); return; } continue; } if (n == 0) { conn->closing = 1; + rpc_trace_end(RPC_TRACE_SERVER_READ, trace); return; } if (errno == EAGAIN || errno == EWOULDBLOCK) { + rpc_trace_end(RPC_TRACE_SERVER_READ, trace); return; } if (errno == EINTR) { continue; } conn->closing = 1; + rpc_trace_end(RPC_TRACE_SERVER_READ, trace); return; } } static void connection_write(rpc_connection *conn) { + uint64_t trace = rpc_trace_begin(); while (conn->write_off < conn->write_len) { ssize_t n = send(conn->fd, conn->write_buf + conn->write_off, conn->write_len - conn->write_off, 0); @@ -306,19 +380,24 @@ static void connection_write(rpc_connection *conn) { continue; } if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + (void)connection_set_interests(conn, RPC_BACKEND_READ | RPC_BACKEND_WRITE); + rpc_trace_end(RPC_TRACE_SERVER_WRITE, trace); return; } conn->closing = 1; + rpc_trace_end(RPC_TRACE_SERVER_WRITE, trace); return; } conn->write_len = 0; conn->write_off = 0; - (void)rpc_backend_modify(conn->server->backend, conn->fd, RPC_BACKEND_READ, - (uintptr_t)conn); + (void)connection_set_interests(conn, RPC_BACKEND_READ); + rpc_trace_end(RPC_TRACE_SERVER_WRITE, trace); } static void accept_ready(rpc_listener *listener) { + uint64_t trace = rpc_trace_begin(); + rpc_worker *worker = listener->worker; for (;;) { struct sockaddr_storage addr; socklen_t addr_len = sizeof(addr); @@ -327,64 +406,129 @@ static void accept_ready(rpc_listener *listener) { if (errno == EINTR) { continue; } + rpc_trace_end(RPC_TRACE_SERVER_ACCEPT, trace); return; } if (set_nonblock(fd) != 0) { close(fd); continue; } - rpc_connection *conn = rpc_fixed_arena_alloc(&listener->server->connection_arena); + rpc_connection *conn = rpc_fixed_arena_alloc(&worker->connection_arena); if (!conn) { close(fd); continue; } conn->fd = fd; - conn->server = listener->server; - conn->next = listener->server->connections; - listener->server->connections = conn; - if (rpc_backend_register(listener->server->backend, fd, RPC_BACKEND_READ, + conn->worker = worker; + conn->next = worker->connections; + worker->connections = conn; + if (rpc_backend_register(worker->backend, fd, RPC_BACKEND_READ, (uintptr_t)conn) != 0) { connection_destroy(conn); + } else { + conn->interests = RPC_BACKEND_READ; } } } -int rpc_server_init(rpc_server **out_server) { - if (!out_server) { +static int worker_init(rpc_server *server, rpc_worker *worker, uint32_t index) { + memset(worker, 0, sizeof(*worker)); + worker->server = server; + worker->index = index; + for (size_t i = 0; i < RPC_MAX_LISTENERS; ++i) { + worker->listeners[i].fd = -1; + } + if (rpc_backend_kqueue_create(&worker->backend) != 0) { return -1; } - rpc_server *server = calloc(1, sizeof(*server)); - if (!server) { + if (rpc_fixed_arena_init(&worker->connection_arena, sizeof(rpc_connection), + RPC_CONNECTION_ARENA_CAPACITY) != 0) { return -1; } - for (size_t i = 0; i < RPC_MAX_LISTENERS; ++i) { - server->listeners[i].fd = -1; + if (rpc_scheduler_init(&worker->scheduler) != 0) { + return -1; } - atomic_init(&server->stopping, false); - if (rpc_backend_kqueue_create(&server->backend) != 0) { - rpc_server_destroy(server); + return 0; +} + +static void worker_destroy(rpc_worker *worker) { + if (!worker) { + return; + } + rpc_connection *conn = worker->connections; + while (conn) { + rpc_connection *next = conn->next; + connection_destroy(conn); + conn = next; + } + for (size_t i = 0; i < worker->listener_count; ++i) { + if (worker->listeners[i].fd >= 0) { + (void)rpc_backend_remove(worker->backend, worker->listeners[i].fd); + close(worker->listeners[i].fd); + worker->listeners[i].fd = -1; + } + } + rpc_scheduler_destroy(worker->scheduler); + rpc_fixed_arena_destroy(&worker->connection_arena); + rpc_backend_destroy(worker->backend); + memset(worker, 0, sizeof(*worker)); +} + +static int server_ensure_workers(rpc_server *server) { + if (server->workers_ready) { + return 0; + } + if (server->worker_count == 0) { + server->worker_count = cpu_count(); + } + server->workers = calloc(server->worker_count, sizeof(*server->workers)); + if (!server->workers) { return -1; } - if (rpc_fixed_arena_init(&server->connection_arena, sizeof(rpc_connection), - RPC_CONNECTION_ARENA_CAPACITY) != 0) { - rpc_server_destroy(server); + for (uint32_t i = 0; i < server->worker_count; ++i) { + if (worker_init(server, &server->workers[i], i) != 0) { + for (uint32_t j = 0; j <= i; ++j) { + worker_destroy(&server->workers[j]); + } + free(server->workers); + server->workers = NULL; + return -1; + } + } + server->workers_ready = 1; + return 0; +} + +int rpc_server_init(rpc_server **out_server) { + if (!out_server) { return -1; } + rpc_server *server = calloc(1, sizeof(*server)); + if (!server) { + return -1; + } + atomic_init(&server->stopping, false); + server->worker_count = cpu_count(); if (rpc_routes_init(&server->routes) != 0) { rpc_server_destroy(server); return -1; } server->routes_ready = 1; - if (rpc_scheduler_init(&server->scheduler) != 0) { - rpc_server_destroy(server); + *out_server = server; + return 0; +} + +int rpc_server_set_workers(rpc_server *server, uint32_t worker_count) { + if (!server || server->workers_ready || server->listening || worker_count == 0 || + worker_count > RPC_MAX_WORKERS) { return -1; } - *out_server = server; + server->worker_count = worker_count; return 0; } int rpc_server_bind(rpc_server *server, const char *host, const char *port) { - if (!server || !port || server->listener_count >= RPC_MAX_LISTENERS) { + if (!server || !port || server_ensure_workers(server) != 0) { return -1; } @@ -401,34 +545,66 @@ int rpc_server_bind(rpc_server *server, const char *host, const char *port) { int ok = -1; for (struct addrinfo *ai = res; ai; ai = ai->ai_next) { - if (server->listener_count >= RPC_MAX_LISTENERS) { + if (server->workers[0].listener_count >= RPC_MAX_LISTENERS) { break; } - int fd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol); - if (fd < 0) { - continue; - } - int yes = 1; - (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); - if (set_nonblock(fd) != 0 || - bind(fd, ai->ai_addr, ai->ai_addrlen) != 0) { - close(fd); - continue; + + size_t added[RPC_MAX_WORKERS]; + memset(added, 0, sizeof(added)); + int bound_all = 1; + + for (uint32_t wi = 0; wi < server->worker_count; ++wi) { + rpc_worker *worker = &server->workers[wi]; + int fd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol); + if (fd < 0) { + bound_all = 0; + break; + } + int yes = 1; + (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); +#ifdef SO_REUSEPORT + (void)setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &yes, sizeof(yes)); +#endif + + struct sockaddr_storage addr; + memcpy(&addr, ai->ai_addr, ai->ai_addrlen); + if (server->port != 0) { + sockaddr_set_port((struct sockaddr *)&addr, server->port); + } + + if (set_nonblock(fd) != 0 || + bind(fd, (struct sockaddr *)&addr, ai->ai_addrlen) != 0) { + close(fd); + bound_all = 0; + break; + } + + if (server->port == 0) { + struct sockaddr_storage bound; + socklen_t bound_len = sizeof(bound); + if (getsockname(fd, (struct sockaddr *)&bound, &bound_len) == 0) { + server->port = sockaddr_get_port((struct sockaddr *)&bound); + } + } + + rpc_listener *listener = &worker->listeners[worker->listener_count++]; + listener->fd = fd; + listener->worker = worker; + added[wi] = 1; } - rpc_listener *listener = &server->listeners[server->listener_count++]; - listener->fd = fd; - listener->server = server; - if (server->port == 0) { - struct sockaddr_storage bound; - socklen_t bound_len = sizeof(bound); - if (getsockname(fd, (struct sockaddr *)&bound, &bound_len) == 0) { - if (bound.ss_family == AF_INET) { - server->port = ntohs(((struct sockaddr_in *)&bound)->sin_port); - } else if (bound.ss_family == AF_INET6) { - server->port = ntohs(((struct sockaddr_in6 *)&bound)->sin6_port); + if (!bound_all) { + for (uint32_t wi = 0; wi < server->worker_count; ++wi) { + if (!added[wi]) { + continue; } + rpc_worker *worker = &server->workers[wi]; + rpc_listener *listener = &worker->listeners[--worker->listener_count]; + close(listener->fd); + listener->fd = -1; + listener->worker = NULL; } + continue; } ok = 0; } @@ -445,31 +621,38 @@ int rpc_server_listen(rpc_server *server) { if (!server) { return -1; } - for (size_t i = 0; i < server->listener_count; ++i) { - rpc_listener *listener = &server->listeners[i]; - if (listen(listener->fd, SOMAXCONN) != 0) { - return -1; - } - uintptr_t user = ((uintptr_t)listener) | 1u; - if (rpc_backend_register(server->backend, listener->fd, RPC_BACKEND_READ, - user) != 0) { - return -1; + for (uint32_t wi = 0; wi < server->worker_count; ++wi) { + rpc_worker *worker = &server->workers[wi]; + for (size_t i = 0; i < worker->listener_count; ++i) { + rpc_listener *listener = &worker->listeners[i]; + if (listen(listener->fd, SOMAXCONN) != 0) { + return -1; + } + uintptr_t user = ((uintptr_t)listener) | 1u; + if (rpc_backend_register(worker->backend, listener->fd, RPC_BACKEND_READ, + user) != 0) { + return -1; + } } } server->listening = 1; return 0; } -int rpc_server_run(rpc_server *server) { - if (!server || !server->listening) { - return -1; - } +static void *worker_run_main(void *arg); + +static int worker_run(rpc_worker *worker) { + rpc_server *server = worker->server; rpc_backend_event events[RPC_EVENT_BATCH]; while (!atomic_load_explicit(&server->stopping, memory_order_acquire)) { - int n = rpc_backend_poll(server->backend, events, RPC_EVENT_BATCH, -1); + uint64_t trace_poll = rpc_trace_begin(); + int n = rpc_backend_poll(worker->backend, events, RPC_EVENT_BATCH, -1); + rpc_trace_end(RPC_TRACE_SERVER_POLL_WAIT, trace_poll); if (n < 0) { return -1; } + rpc_trace_add(RPC_TRACE_SERVER_POLL_EVENTS, (uint64_t)n); + uint64_t trace_active = rpc_trace_begin(); for (int i = 0; i < n; ++i) { if (events[i].events & RPC_BACKEND_WAKE) { continue; @@ -488,7 +671,44 @@ int rpc_server_run(rpc_server *server) { maybe_close(conn); } } - rpc_scheduler_run_ready(server->scheduler); + uint64_t trace_schedule = rpc_trace_begin(); + rpc_scheduler_run_ready(worker->scheduler); + rpc_trace_end(RPC_TRACE_SERVER_SCHEDULE, trace_schedule); + rpc_trace_end(RPC_TRACE_SERVER_LOOP_ACTIVE, trace_active); + } + return 0; +} + +static void *worker_run_main(void *arg) { + (void)worker_run(arg); + return NULL; +} + +int rpc_server_run(rpc_server *server) { + if (!server || !server->listening) { + return -1; + } + + if (server->worker_count == 1) { + return worker_run(&server->workers[0]); + } + + for (uint32_t i = 0; i < server->worker_count; ++i) { + if (pthread_create(&server->workers[i].thread, NULL, worker_run_main, + &server->workers[i]) != 0) { + rpc_server_stop(server); + for (uint32_t j = 0; j < i; ++j) { + pthread_join(server->workers[j].thread, NULL); + server->workers[j].thread_started = 0; + } + return -1; + } + server->workers[i].thread_started = 1; + } + + for (uint32_t i = 0; i < server->worker_count; ++i) { + pthread_join(server->workers[i].thread, NULL); + server->workers[i].thread_started = 0; } return 0; } @@ -498,31 +718,31 @@ void rpc_server_stop(rpc_server *server) { return; } atomic_store_explicit(&server->stopping, true, memory_order_release); - (void)rpc_backend_wake(server->backend); + for (uint32_t i = 0; i < server->worker_count; ++i) { + if (server->workers && server->workers[i].backend) { + (void)rpc_backend_wake(server->workers[i].backend); + } + } } void rpc_server_destroy(rpc_server *server) { if (!server) { return; } - rpc_connection *conn = server->connections; - while (conn) { - rpc_connection *next = conn->next; - connection_destroy(conn); - conn = next; - } - for (size_t i = 0; i < server->listener_count; ++i) { - if (server->listeners[i].fd >= 0) { - (void)rpc_backend_remove(server->backend, server->listeners[i].fd); - close(server->listeners[i].fd); + if (server->workers) { + rpc_server_stop(server); + for (uint32_t i = 0; i < server->worker_count; ++i) { + if (server->workers[i].thread_started) { + pthread_join(server->workers[i].thread, NULL); + server->workers[i].thread_started = 0; + } + worker_destroy(&server->workers[i]); } + free(server->workers); } - rpc_scheduler_destroy(server->scheduler); if (server->routes_ready) { rpc_routes_destroy(&server->routes); } - rpc_fixed_arena_destroy(&server->connection_arena); - rpc_backend_destroy(server->backend); free(server); } diff --git a/src/trace.c b/src/trace.c new file mode 100644 index 0000000..05ad4f5 --- /dev/null +++ b/src/trace.c @@ -0,0 +1,175 @@ +#include "rpc/trace.h" + +#include +#include + +typedef struct rpc_trace_counter { + atomic_uint_fast64_t count; + atomic_uint_fast64_t total_ns; + atomic_uint_fast64_t max_ns; +} rpc_trace_counter; + +static const char *trace_names[RPC_TRACE_COUNT] = { + [RPC_TRACE_CLIENT_CALL] = "client.call", + [RPC_TRACE_CLIENT_SEND] = "client.send", + [RPC_TRACE_CLIENT_RECV] = "client.recv", + [RPC_TRACE_CLIENT_DECODE] = "client.decode", + [RPC_TRACE_SERVER_ACCEPT] = "server.accept", + [RPC_TRACE_SERVER_POLL_WAIT] = "server.poll_wait", + [RPC_TRACE_SERVER_POLL_EVENTS] = "server.poll_events", + [RPC_TRACE_SERVER_LOOP_ACTIVE] = "server.loop_active", + [RPC_TRACE_SERVER_READ] = "server.read", + [RPC_TRACE_SERVER_PARSE] = "server.parse", + [RPC_TRACE_SERVER_ROUTE] = "server.route", + [RPC_TRACE_SERVER_SCHEDULE] = "server.schedule", + [RPC_TRACE_SERVER_WRITE] = "server.write", + [RPC_TRACE_SCHED_SUBMIT] = "sched.submit", + [RPC_TRACE_SCHED_DECODE] = "sched.decode", + [RPC_TRACE_SCHED_CORO_CREATE] = "sched.coro_create", + [RPC_TRACE_SCHED_RESUME] = "sched.resume", + [RPC_TRACE_PAYLOAD_DECODE] = "payload.decode", + [RPC_TRACE_ROUTE_LOOKUP] = "route.lookup", +}; + +static const int trace_is_time[RPC_TRACE_COUNT] = { + [RPC_TRACE_CLIENT_CALL] = 1, + [RPC_TRACE_CLIENT_SEND] = 1, + [RPC_TRACE_CLIENT_RECV] = 1, + [RPC_TRACE_CLIENT_DECODE] = 1, + [RPC_TRACE_SERVER_ACCEPT] = 1, + [RPC_TRACE_SERVER_POLL_WAIT] = 1, + [RPC_TRACE_SERVER_LOOP_ACTIVE] = 1, + [RPC_TRACE_SERVER_READ] = 1, + [RPC_TRACE_SERVER_PARSE] = 1, + [RPC_TRACE_SERVER_ROUTE] = 1, + [RPC_TRACE_SERVER_SCHEDULE] = 1, + [RPC_TRACE_SERVER_WRITE] = 1, + [RPC_TRACE_SCHED_SUBMIT] = 1, + [RPC_TRACE_SCHED_DECODE] = 1, + [RPC_TRACE_SCHED_CORO_CREATE] = 1, + [RPC_TRACE_SCHED_RESUME] = 1, + [RPC_TRACE_PAYLOAD_DECODE] = 1, + [RPC_TRACE_ROUTE_LOOKUP] = 1, +}; + +static const char *trace_avg_units[RPC_TRACE_COUNT] = { + [RPC_TRACE_SERVER_POLL_EVENTS] = "events/poll", +}; + +static const char *trace_max_units[RPC_TRACE_COUNT] = { + [RPC_TRACE_SERVER_POLL_EVENTS] = "events", +}; + +static rpc_trace_counter counters[RPC_TRACE_COUNT]; + +int rpc_trace_enabled = 0; + +static void format_duration(uint64_t ns, char out[16]) { + if (ns < 1000ull) { + snprintf(out, 16, "%llu ns", (unsigned long long)ns); + } else if (ns < 1000000ull) { + snprintf(out, 16, "%.2f us", (double)ns / 1000.0); + } else if (ns < 1000000000ull) { + snprintf(out, 16, "%.2f ms", (double)ns / 1000000.0); + } else { + snprintf(out, 16, "%.3f s", (double)ns / 1000000000.0); + } +} + +void rpc_trace_set_enabled(int enabled) { +#if defined(__GNUC__) || defined(__clang__) + __atomic_store_n(&rpc_trace_enabled, enabled ? 1 : 0, __ATOMIC_RELAXED); +#else + rpc_trace_enabled = enabled ? 1 : 0; +#endif +} + +uint64_t rpc_trace_begin_slow(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; +} + +void rpc_trace_end_slow(rpc_trace_metric metric, uint64_t start_ns) { + rpc_trace_add_slow(metric, rpc_trace_begin_slow() - start_ns); +} + +void rpc_trace_add_slow(rpc_trace_metric metric, uint64_t value) { + if ((unsigned)metric >= RPC_TRACE_COUNT) { + return; + } + rpc_trace_counter *counter = &counters[metric]; + atomic_fetch_add_explicit(&counter->count, 1, memory_order_relaxed); + atomic_fetch_add_explicit(&counter->total_ns, value, memory_order_relaxed); + + uint64_t old = atomic_load_explicit(&counter->max_ns, memory_order_relaxed); + while (old < value && + !atomic_compare_exchange_weak_explicit(&counter->max_ns, &old, value, + memory_order_relaxed, + memory_order_relaxed)) { + } +} + +void rpc_trace_reset(void) { + for (size_t i = 0; i < RPC_TRACE_COUNT; ++i) { + atomic_store_explicit(&counters[i].count, 0, memory_order_relaxed); + atomic_store_explicit(&counters[i].total_ns, 0, memory_order_relaxed); + atomic_store_explicit(&counters[i].max_ns, 0, memory_order_relaxed); + } +} + +void rpc_trace_snapshot(rpc_trace_stat out[RPC_TRACE_COUNT]) { + for (size_t i = 0; i < RPC_TRACE_COUNT; ++i) { + out[i] = (rpc_trace_stat){ + .name = trace_names[i], + .count = atomic_load_explicit(&counters[i].count, memory_order_relaxed), + .total = + atomic_load_explicit(&counters[i].total_ns, memory_order_relaxed), + .max = + atomic_load_explicit(&counters[i].max_ns, memory_order_relaxed), + .is_time = trace_is_time[i], + }; + } +} + +void rpc_trace_dump(FILE *out) { + rpc_trace_stat stats[RPC_TRACE_COUNT]; + rpc_trace_snapshot(stats); + if (!out) { + out = stderr; + } + + fprintf(out, "\ntrace:\n"); + fprintf(out, " %-22s %12s %12s %12s\n", "metric", "count", "avg", + "max"); + for (size_t i = 0; i < RPC_TRACE_COUNT; ++i) { + if (stats[i].count == 0) { + continue; + } + if (stats[i].is_time) { + char avg[16]; + char max[16]; + format_duration(stats[i].total / stats[i].count, avg); + format_duration(stats[i].max, max); + fprintf(out, " %-22s %12llu %12s %12s\n", stats[i].name, + (unsigned long long)stats[i].count, avg, max); + } else { + double avg = (double)stats[i].total / (double)stats[i].count; + const char *avg_unit = trace_avg_units[i]; + const char *max_unit = trace_max_units[i]; + if (avg_unit && max_unit) { + char avg_buf[24]; + char max_buf[24]; + snprintf(avg_buf, sizeof(avg_buf), "%.2f %s", avg, avg_unit); + snprintf(max_buf, sizeof(max_buf), "%llu %s", + (unsigned long long)stats[i].max, max_unit); + fprintf(out, " %-22s %12llu %12s %12s\n", stats[i].name, + (unsigned long long)stats[i].count, avg_buf, max_buf); + } else { + fprintf(out, " %-22s %12llu %12.2f %12llu\n", stats[i].name, + (unsigned long long)stats[i].count, avg, + (unsigned long long)stats[i].max); + } + } + } +}