diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/base64.c | 56 | ||||
| -rw-r--r-- | src/buf.c | 30 | ||||
| -rw-r--r-- | src/ci.c | 76 | ||||
| -rw-r--r-- | src/client.c | 122 | ||||
| -rw-r--r-- | src/client.h | 4 | ||||
| -rw-r--r-- | src/client_main.c | 94 | ||||
| -rw-r--r-- | src/cmd_line.c | 52 | ||||
| -rw-r--r-- | src/command.c | 228 | ||||
| -rw-r--r-- | src/command.h | 8 | ||||
| -rw-r--r-- | src/event_loop.c | 260 | ||||
| -rw-r--r-- | src/file.c | 176 | ||||
| -rw-r--r-- | src/git.c | 126 | ||||
| -rw-r--r-- | src/json.c | 284 | ||||
| -rw-r--r-- | src/json_rpc.c | 568 | ||||
| -rw-r--r-- | src/log.c | 42 | ||||
| -rw-r--r-- | src/log.h | 102 | ||||
| -rw-r--r-- | src/net.c | 315 | ||||
| -rw-r--r-- | src/process.c | 184 | ||||
| -rw-r--r-- | src/process.h | 6 | ||||
| -rw-r--r-- | src/protocol.c | 230 | ||||
| -rw-r--r-- | src/run_queue.c | 192 | ||||
| -rw-r--r-- | src/run_queue.h | 4 | ||||
| -rw-r--r-- | src/server.c | 537 | ||||
| -rw-r--r-- | src/server.h | 4 | ||||
| -rw-r--r-- | src/server_main.c | 94 | ||||
| -rw-r--r-- | src/signal.c | 66 | ||||
| -rw-r--r-- | src/sqlite.c | 338 | ||||
| -rw-r--r-- | src/storage.c | 110 | ||||
| -rw-r--r-- | src/storage.h | 18 | ||||
| -rw-r--r-- | src/storage_sqlite.c | 648 | ||||
| -rw-r--r-- | src/string.c | 44 | ||||
| -rw-r--r-- | src/tcp_server.c | 246 | ||||
| -rw-r--r-- | src/worker.c | 287 | ||||
| -rw-r--r-- | src/worker.h | 4 | ||||
| -rw-r--r-- | src/worker_main.c | 94 | ||||
| -rw-r--r-- | src/worker_queue.c | 54 |
36 files changed, 2849 insertions, 2854 deletions
diff --git a/src/base64.c b/src/base64.c index 733c700..0192eca 100644 --- a/src/base64.c +++ b/src/base64.c @@ -18,44 +18,44 @@ static const int base64_variant = sodium_base64_VARIANT_ORIGINAL; int base64_encode(const unsigned char* src, size_t src_len, char** _dst) { - const size_t dst_len = sodium_base64_encoded_len(src_len, base64_variant); + const size_t dst_len = sodium_base64_encoded_len(src_len, base64_variant); - char* dst = calloc(dst_len, 1); - if (!dst) { - log_errno("calloc"); - return -1; - } + char* dst = calloc(dst_len, 1); + if (!dst) { + log_errno("calloc"); + return -1; + } - sodium_bin2base64(dst, dst_len, src, src_len, base64_variant); + sodium_bin2base64(dst, dst_len, src, src_len, base64_variant); - *_dst = dst; - return 0; + *_dst = dst; + return 0; } int base64_decode(const char* src, unsigned char** _dst, size_t* _dst_len) { - const size_t src_len = strlen(src); - const size_t dst_max_len = src_len / 4 * 3; - size_t dst_len = 0; + const size_t src_len = strlen(src); + const size_t dst_max_len = src_len / 4 * 3; + size_t dst_len = 0; - unsigned char* dst = calloc(dst_max_len, 1); - if (!dst) { - log_errno("calloc"); - return -1; - } + unsigned char* dst = calloc(dst_max_len, 1); + if (!dst) { + log_errno("calloc"); + return -1; + } - int ret = - sodium_base642bin(dst, dst_max_len, src, src_len, NULL, &dst_len, NULL, base64_variant); - if (ret < 0) { - log_err("Couldn't parse base64-encoded string\n"); - goto free; - } + int ret = + sodium_base642bin(dst, dst_max_len, src, src_len, NULL, &dst_len, NULL, base64_variant); + if (ret < 0) { + log_err("Couldn't parse base64-encoded string\n"); + goto free; + } - *_dst = dst; - *_dst_len = dst_len; - return ret; + *_dst = dst; + *_dst_len = dst_len; + return ret; free: - free(dst); + free(dst); - return ret; + return ret; } @@ -14,36 +14,36 @@ #include <string.h> struct buf { - uint32_t size; - const void* data; + uint32_t size; + const void* data; }; int buf_create(struct buf** _buf, const void* data, uint32_t size) { - struct buf* buf = malloc(sizeof(struct buf)); - if (!buf) { - log_errno("malloc"); - return -1; - } + struct buf* buf = malloc(sizeof(struct buf)); + if (!buf) { + log_errno("malloc"); + return -1; + } - buf->data = data; - buf->size = size; + buf->data = data; + buf->size = size; - *_buf = buf; - return 0; + *_buf = buf; + return 0; } int buf_create_from_string(struct buf** buf, const char* str) { - return buf_create(buf, str, strlen(str) + 1); + return buf_create(buf, str, strlen(str) + 1); } void buf_destroy(struct buf* buf) { - free(buf); + free(buf); } uint32_t buf_get_size(const struct buf* buf) { - return buf->size; + return buf->size; } const void* buf_get_data(const struct buf* buf) { - return buf->data; + return buf->data; } @@ -34,70 +34,70 @@ static const char *ci_env[] = { /* clang-format on */ static int ci_run_script(const char* script, struct process_output* result) { - const char* args[] = {script, NULL}; - return process_execute_and_capture(args, ci_env, result); + const char* args[] = {script, NULL}; + return process_execute_and_capture(args, ci_env, result); } int ci_run(struct process_output* result) { - for (const char** script = ci_scripts; *script; ++script) { - if (!file_exists(*script)) - continue; - log("Going to run: %s\n", *script); - return ci_run_script(*script, result); - } + for (const char** script = ci_scripts; *script; ++script) { + if (!file_exists(*script)) + continue; + log("Going to run: %s\n", *script); + return ci_run_script(*script, result); + } - log("Couldn't find any CI scripts to run\n"); - return -1; + log("Couldn't find any CI scripts to run\n"); + return -1; } static void ci_cleanup_git_repo(git_repository* repo) { - rm_rf(git_repository_workdir(repo)); - libgit_repository_free(repo); + rm_rf(git_repository_workdir(repo)); + libgit_repository_free(repo); } static int ci_prepare_git_repo(git_repository** repo, const char* url, const char* rev) { - int ret = 0; + int ret = 0; - ret = libgit_clone_to_tmp(repo, url); - if (ret < 0) - return ret; + ret = libgit_clone_to_tmp(repo, url); + if (ret < 0) + return ret; - ret = libgit_checkout(*repo, rev); - if (ret < 0) - goto cleanup_repo; + ret = libgit_checkout(*repo, rev); + if (ret < 0) + goto cleanup_repo; - return ret; + return ret; cleanup_repo: - ci_cleanup_git_repo(*repo); + ci_cleanup_git_repo(*repo); - return ret; + return ret; } int ci_run_git_repo(const char* url, const char* rev, struct process_output* output) { - char* oldpwd = NULL; - git_repository* repo = NULL; - int ret = 0; + char* oldpwd = NULL; + git_repository* repo = NULL; + int ret = 0; - ret = ci_prepare_git_repo(&repo, url, rev); - if (ret < 0) - goto exit; + ret = ci_prepare_git_repo(&repo, url, rev); + if (ret < 0) + goto exit; - ret = chdir_wrapper(git_repository_workdir(repo), &oldpwd); - if (ret < 0) - goto free_repo; + ret = chdir_wrapper(git_repository_workdir(repo), &oldpwd); + if (ret < 0) + goto free_repo; - ret = ci_run(output); - if (ret < 0) - goto oldpwd; + ret = ci_run(output); + if (ret < 0) + goto oldpwd; oldpwd: - chdir_wrapper(oldpwd, NULL); - free(oldpwd); + chdir_wrapper(oldpwd, NULL); + free(oldpwd); free_repo: - ci_cleanup_git_repo(repo); + ci_cleanup_git_repo(repo); exit: - return ret; + return ret; } diff --git a/src/client.c b/src/client.c index d703baa..493dec2 100644 --- a/src/client.c +++ b/src/client.c @@ -20,98 +20,98 @@ #include <string.h> struct client { - int dummy; + int dummy; }; int client_create(struct client** _client) { - int ret = 0; + int ret = 0; - struct client* client = malloc(sizeof(struct client)); - if (!client) { - log_errno("malloc"); - return -1; - } + struct client* client = malloc(sizeof(struct client)); + if (!client) { + log_errno("malloc"); + return -1; + } - *_client = client; - return ret; + *_client = client; + return ret; } void client_destroy(struct client* client) { - free(client); + free(client); } static int make_request(struct jsonrpc_request** request, int argc, const char** argv) { - if (argc < 1) { - exit_with_usage_err("no action specified"); - return -1; - } + if (argc < 1) { + exit_with_usage_err("no action specified"); + return -1; + } - if (!strcmp(argv[0], CMD_QUEUE_RUN)) { - if (argc != 3) - return -1; + if (!strcmp(argv[0], CMD_QUEUE_RUN)) { + if (argc != 3) + return -1; - struct run* run = NULL; - int ret = run_queued(&run, argv[1], argv[2]); - if (ret < 0) - return ret; + struct run* run = NULL; + int ret = run_queued(&run, argv[1], argv[2]); + if (ret < 0) + return ret; - ret = request_create_queue_run(request, run); - run_destroy(run); - return ret; - } else if (!strcmp(argv[0], CMD_GET_RUNS)) { - if (argc != 1) - return -1; - return request_create_get_runs(request); - } + ret = request_create_queue_run(request, run); + run_destroy(run); + return ret; + } else if (!strcmp(argv[0], CMD_GET_RUNS)) { + if (argc != 1) + return -1; + return request_create_get_runs(request); + } - return -1; + return -1; } int client_main(UNUSED const struct client* client, const struct settings* settings, int argc, const char** argv) { - int ret = 0; + int ret = 0; - struct jsonrpc_request* request = NULL; - ret = make_request(&request, argc, argv); - if (ret < 0) { - exit_with_usage_err("invalid request"); - return ret; - } + struct jsonrpc_request* request = NULL; + ret = make_request(&request, argc, argv); + if (ret < 0) { + exit_with_usage_err("invalid request"); + return ret; + } - ret = net_connect(settings->host, settings->port); - if (ret < 0) - goto free_request; - int fd = ret; + ret = net_connect(settings->host, settings->port); + if (ret < 0) + goto free_request; + int fd = ret; - ret = jsonrpc_request_send(request, fd); - if (ret < 0) - goto close; + ret = jsonrpc_request_send(request, fd); + if (ret < 0) + goto close; - struct jsonrpc_response* response = NULL; - ret = jsonrpc_response_recv(&response, fd); - if (ret < 0) - goto close; + struct jsonrpc_response* response = NULL; + ret = jsonrpc_response_recv(&response, fd); + if (ret < 0) + goto close; - const char* response_str = jsonrpc_response_to_string(response); - if (response_str) { - if (jsonrpc_response_is_error(response)) - ret = -1; - printf("%s", response_str); - } else { - log_err("no response\n"); - } - goto free_response; + const char* response_str = jsonrpc_response_to_string(response); + if (response_str) { + if (jsonrpc_response_is_error(response)) + ret = -1; + printf("%s", response_str); + } else { + log_err("no response\n"); + } + goto free_response; free_response: - jsonrpc_response_destroy(response); + jsonrpc_response_destroy(response); close: - net_close(fd); + net_close(fd); free_request: - jsonrpc_request_destroy(request); + jsonrpc_request_destroy(request); - return ret; + return ret; } diff --git a/src/client.h b/src/client.h index 5df5561..a00890f 100644 --- a/src/client.h +++ b/src/client.h @@ -9,8 +9,8 @@ #define __CLIENT_H__ struct settings { - const char* host; - const char* port; + const char* host; + const char* port; }; struct client; diff --git a/src/client_main.c b/src/client_main.c index 863cf9d..6419157 100644 --- a/src/client_main.c +++ b/src/client_main.c @@ -14,26 +14,26 @@ #include <unistd.h> static struct settings default_settings(void) { - struct settings settings = { - .host = default_host, - .port = default_port, - }; - return settings; + struct settings settings = { + .host = default_host, + .port = default_port, + }; + return settings; } const char* get_usage_string(void) { - return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT] ACTION [ARG...]\n\ + return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT] ACTION [ARG...]\n\ \n\ available actions:\n\ \t" CMD_QUEUE_RUN " URL REV - schedule a CI run of repository at URL, revision REV"; } static int parse_settings(struct settings* settings, int argc, char* argv[]) { - int opt, longind; + int opt, longind; - *settings = default_settings(); + *settings = default_settings(); - /* clang-format off */ + /* clang-format off */ static struct option long_options[] = { {"help", no_argument, 0, 'h'}, {"version", no_argument, 0, 'V'}, @@ -42,53 +42,53 @@ static int parse_settings(struct settings* settings, int argc, char* argv[]) { {"port", required_argument, 0, 'p'}, {0, 0, 0, 0}, }; - /* clang-format on */ + /* clang-format on */ - while ((opt = getopt_long(argc, argv, "hVvH:p:", long_options, &longind)) != -1) { - switch (opt) { - case 'h': - exit_with_usage(0); - break; - case 'V': - exit_with_version(); - break; - case 'v': - g_log_lvl = LOG_LVL_DEBUG; - break; - case 'H': - settings->host = optarg; - break; - case 'p': - settings->port = optarg; - break; - default: - exit_with_usage(1); - break; - } - } + while ((opt = getopt_long(argc, argv, "hVvH:p:", long_options, &longind)) != -1) { + switch (opt) { + case 'h': + exit_with_usage(0); + break; + case 'V': + exit_with_version(); + break; + case 'v': + g_log_lvl = LOG_LVL_DEBUG; + break; + case 'H': + settings->host = optarg; + break; + case 'p': + settings->port = optarg; + break; + default: + exit_with_usage(1); + break; + } + } - return 0; + return 0; } int main(int argc, char* argv[]) { - struct settings settings; - struct client* client = NULL; - int ret = 0; + struct settings settings; + struct client* client = NULL; + int ret = 0; - ret = parse_settings(&settings, argc, argv); - if (ret < 0) - return ret; + ret = parse_settings(&settings, argc, argv); + if (ret < 0) + return ret; - ret = client_create(&client); - if (ret < 0) - return ret; + ret = client_create(&client); + if (ret < 0) + return ret; - ret = client_main(client, &settings, argc - optind, (const char**)argv + optind); - if (ret < 0) - goto destroy_client; + ret = client_main(client, &settings, argc - optind, (const char**)argv + optind); + if (ret < 0) + goto destroy_client; destroy_client: - client_destroy(client); + client_destroy(client); - return ret; + return ret; } diff --git a/src/cmd_line.c b/src/cmd_line.c index be1a947..db6b36b 100644 --- a/src/cmd_line.c +++ b/src/cmd_line.c @@ -16,50 +16,50 @@ #include <string.h> static char* get_current_binary_path(void) { - return readlink_wrapper("/proc/self/exe"); + return readlink_wrapper("/proc/self/exe"); } static char* get_current_binary_name(void) { - char* path = get_current_binary_path(); - if (!path) - return NULL; + char* path = get_current_binary_path(); + if (!path) + return NULL; - char* name = basename(path); + char* name = basename(path); - char* result = strdup(name); - if (!result) { - log_errno("strdup"); - goto free_path; - } + char* result = strdup(name); + if (!result) { + log_errno("strdup"); + goto free_path; + } free_path: - free(path); + free(path); - return result; + return result; } void exit_with_usage(int ec) { - FILE* dest = stdout; - if (ec) - dest = stderr; + FILE* dest = stdout; + if (ec) + dest = stderr; - char* binary = get_current_binary_name(); + char* binary = get_current_binary_name(); - fprintf(dest, "usage: %s %s\n", binary ? binary : "prog", get_usage_string()); - free(binary); - exit(ec); + fprintf(dest, "usage: %s %s\n", binary ? binary : "prog", get_usage_string()); + free(binary); + exit(ec); } void exit_with_usage_err(const char* msg) { - if (msg) - fprintf(stderr, "usage error: %s\n", msg); - exit_with_usage(1); + if (msg) + fprintf(stderr, "usage error: %s\n", msg); + exit_with_usage(1); } void exit_with_version(void) { - char* binary = get_current_binary_name(); + char* binary = get_current_binary_name(); - printf("%s v%s (%s)\n", binary ? binary : "prog", project_version, project_rev); - free(binary); - exit(0); + printf("%s v%s (%s)\n", binary ? binary : "prog", project_version, project_rev); + free(binary); + exit(0); } diff --git a/src/command.c b/src/command.c index bba4ace..18e15c6 100644 --- a/src/command.c +++ b/src/command.c @@ -17,203 +17,203 @@ #include <string.h> struct cmd_dispatcher { - struct cmd_desc* cmds; - size_t numof_cmds; - void* ctx; + struct cmd_desc* cmds; + size_t numof_cmds; + void* ctx; }; static int copy_cmd(struct cmd_desc* dest, const struct cmd_desc* src) { - dest->name = strdup(src->name); - if (!dest->name) { - log_errno("strdup"); - return -1; - } - dest->handler = src->handler; - return 0; + dest->name = strdup(src->name); + if (!dest->name) { + log_errno("strdup"); + return -1; + } + dest->handler = src->handler; + return 0; } static void free_cmd(struct cmd_desc* desc) { - free(desc->name); + free(desc->name); } static int copy_cmds(struct cmd_desc* dest, const struct cmd_desc* src, size_t numof_cmds) { - size_t numof_copied = 0; - int ret = 0; + size_t numof_copied = 0; + int ret = 0; - for (numof_copied = 0; numof_copied < numof_cmds; ++numof_copied) { - ret = copy_cmd(&dest[numof_copied], &src[numof_copied]); - if (ret < 0) - goto free; - } + for (numof_copied = 0; numof_copied < numof_cmds; ++numof_copied) { + ret = copy_cmd(&dest[numof_copied], &src[numof_copied]); + if (ret < 0) + goto free; + } - return 0; + return 0; free: - for (size_t i = 0; i < numof_copied; ++i) - free_cmd(&dest[numof_copied]); + for (size_t i = 0; i < numof_copied; ++i) + free_cmd(&dest[numof_copied]); - return -1; + return -1; } static void free_cmds(struct cmd_desc* cmds, size_t numof_cmds) { - for (size_t i = 0; i < numof_cmds; ++i) - free_cmd(&cmds[i]); + for (size_t i = 0; i < numof_cmds; ++i) + free_cmd(&cmds[i]); } int cmd_dispatcher_create(struct cmd_dispatcher** _dispatcher, struct cmd_desc* cmds, size_t numof_cmds, void* ctx) { - int ret = 0; + int ret = 0; - struct cmd_dispatcher* dispatcher = malloc(sizeof(struct cmd_dispatcher)); - if (!dispatcher) { - log_errno("malloc"); - return -1; - } + struct cmd_dispatcher* dispatcher = malloc(sizeof(struct cmd_dispatcher)); + if (!dispatcher) { + log_errno("malloc"); + return -1; + } - dispatcher->ctx = ctx; + dispatcher->ctx = ctx; - dispatcher->cmds = malloc(sizeof(struct cmd_desc) * numof_cmds); - if (!dispatcher->cmds) { - log_errno("malloc"); - goto free; - } - dispatcher->numof_cmds = numof_cmds; + dispatcher->cmds = malloc(sizeof(struct cmd_desc) * numof_cmds); + if (!dispatcher->cmds) { + log_errno("malloc"); + goto free; + } + dispatcher->numof_cmds = numof_cmds; - ret = copy_cmds(dispatcher->cmds, cmds, numof_cmds); - if (ret < 0) - goto free_cmds; + ret = copy_cmds(dispatcher->cmds, cmds, numof_cmds); + if (ret < 0) + goto free_cmds; - *_dispatcher = dispatcher; - return 0; + *_dispatcher = dispatcher; + return 0; free_cmds: - free(dispatcher->cmds); + free(dispatcher->cmds); free: - free(dispatcher); + free(dispatcher); - return -1; + return -1; } void cmd_dispatcher_destroy(struct cmd_dispatcher* dispatcher) { - free_cmds(dispatcher->cmds, dispatcher->numof_cmds); - free(dispatcher->cmds); - free(dispatcher); + free_cmds(dispatcher->cmds, dispatcher->numof_cmds); + free(dispatcher->cmds); + free(dispatcher); } static int cmd_dispatcher_handle_internal(const struct cmd_dispatcher* dispatcher, const struct jsonrpc_request* request, struct jsonrpc_response** result, void* arg) { - const char* actual_cmd = jsonrpc_request_get_method(request); + const char* actual_cmd = jsonrpc_request_get_method(request); - for (size_t i = 0; i < dispatcher->numof_cmds; ++i) { - struct cmd_desc* cmd = &dispatcher->cmds[i]; + for (size_t i = 0; i < dispatcher->numof_cmds; ++i) { + struct cmd_desc* cmd = &dispatcher->cmds[i]; - if (strcmp(cmd->name, actual_cmd)) - continue; + if (strcmp(cmd->name, actual_cmd)) + continue; - return cmd->handler(request, result, arg); - } + return cmd->handler(request, result, arg); + } - log_err("Received an unknown command: %s\n", actual_cmd); - return -1; + log_err("Received an unknown command: %s\n", actual_cmd); + return -1; } int cmd_dispatcher_handle(const struct cmd_dispatcher* dispatcher, const struct jsonrpc_request* command, struct jsonrpc_response** result) { - return cmd_dispatcher_handle_internal(dispatcher, command, result, dispatcher->ctx); + return cmd_dispatcher_handle_internal(dispatcher, command, result, dispatcher->ctx); } static struct cmd_conn_ctx* make_conn_ctx(int fd, void* arg) { - struct cmd_conn_ctx* ctx = malloc(sizeof(struct cmd_conn_ctx)); - if (!ctx) { - log_errno("malloc"); - return NULL; - } + struct cmd_conn_ctx* ctx = malloc(sizeof(struct cmd_conn_ctx)); + if (!ctx) { + log_errno("malloc"); + return NULL; + } - ctx->fd = fd; - ctx->arg = arg; + ctx->fd = fd; + ctx->arg = arg; - return ctx; + return ctx; } static int cmd_dispatcher_handle_conn_internal(int conn_fd, struct cmd_dispatcher* dispatcher) { - int ret = 0; + int ret = 0; - struct cmd_conn_ctx* new_ctx = make_conn_ctx(conn_fd, dispatcher->ctx); - if (!new_ctx) - return -1; + struct cmd_conn_ctx* new_ctx = make_conn_ctx(conn_fd, dispatcher->ctx); + if (!new_ctx) + return -1; - struct jsonrpc_request* request = NULL; - ret = jsonrpc_request_recv(&request, conn_fd); - if (ret < 0) - goto free_ctx; + struct jsonrpc_request* request = NULL; + ret = jsonrpc_request_recv(&request, conn_fd); + if (ret < 0) + goto free_ctx; - const int requires_response = !jsonrpc_request_is_notification(request); + const int requires_response = !jsonrpc_request_is_notification(request); - struct jsonrpc_response* default_response = NULL; - if (requires_response) { - ret = jsonrpc_response_create(&default_response, request, NULL); - if (ret < 0) - goto free_request; - } + struct jsonrpc_response* default_response = NULL; + if (requires_response) { + ret = jsonrpc_response_create(&default_response, request, NULL); + if (ret < 0) + goto free_request; + } - struct jsonrpc_response* default_error = NULL; - if (requires_response) { - ret = jsonrpc_error_create(&default_error, request, -1, "An error occured"); - if (ret < 0) - goto free_default_response; - } + struct jsonrpc_response* default_error = NULL; + if (requires_response) { + ret = jsonrpc_error_create(&default_error, request, -1, "An error occured"); + if (ret < 0) + goto free_default_response; + } - struct jsonrpc_response* response = NULL; - ret = cmd_dispatcher_handle_internal(dispatcher, request, &response, new_ctx); + struct jsonrpc_response* response = NULL; + ret = cmd_dispatcher_handle_internal(dispatcher, request, &response, new_ctx); - if (requires_response) { - struct jsonrpc_response* actual_response = response; - if (!actual_response) { - actual_response = ret < 0 ? default_error : default_response; - } - if (ret < 0 && !jsonrpc_response_is_error(actual_response)) { - actual_response = default_error; - } - ret = jsonrpc_response_send(actual_response, conn_fd) < 0 ? -1 : ret; - } + if (requires_response) { + struct jsonrpc_response* actual_response = response; + if (!actual_response) { + actual_response = ret < 0 ? default_error : default_response; + } + if (ret < 0 && !jsonrpc_response_is_error(actual_response)) { + actual_response = default_error; + } + ret = jsonrpc_response_send(actual_response, conn_fd) < 0 ? -1 : ret; + } - if (response) - jsonrpc_response_destroy(response); + if (response) + jsonrpc_response_destroy(response); - if (default_error) - jsonrpc_response_destroy(default_error); + if (default_error) + jsonrpc_response_destroy(default_error); free_default_response: - if (default_response) - jsonrpc_response_destroy(default_response); + if (default_response) + jsonrpc_response_destroy(default_response); free_request: - jsonrpc_request_destroy(request); + jsonrpc_request_destroy(request); free_ctx: - free(new_ctx); + free(new_ctx); - return ret; + return ret; } int cmd_dispatcher_handle_conn(int conn_fd, void* _dispatcher) { - return cmd_dispatcher_handle_conn_internal(conn_fd, (struct cmd_dispatcher*)_dispatcher); + return cmd_dispatcher_handle_conn_internal(conn_fd, (struct cmd_dispatcher*)_dispatcher); } int cmd_dispatcher_handle_event(UNUSED struct event_loop* loop, int fd, short revents, void* _dispatcher) { - if (!(revents & POLLIN)) { - log_err("Descriptor %d is not readable\n", fd); - return -1; - } + if (!(revents & POLLIN)) { + log_err("Descriptor %d is not readable\n", fd); + return -1; + } - return cmd_dispatcher_handle_conn_internal(fd, (struct cmd_dispatcher*)_dispatcher); + return cmd_dispatcher_handle_conn_internal(fd, (struct cmd_dispatcher*)_dispatcher); } diff --git a/src/command.h b/src/command.h index fea3118..da7ab24 100644 --- a/src/command.h +++ b/src/command.h @@ -18,8 +18,8 @@ typedef int (*cmd_handler)(const struct jsonrpc_request* request, void* ctx); struct cmd_desc { - char* name; - cmd_handler handler; + char* name; + cmd_handler handler; }; struct cmd_dispatcher; @@ -32,8 +32,8 @@ int cmd_dispatcher_handle(const struct cmd_dispatcher*, struct jsonrpc_response** response); struct cmd_conn_ctx { - int fd; - void* arg; + int fd; + void* arg; }; /* This is supposed to be used as an argument to tcp_server_accept. */ diff --git a/src/event_loop.c b/src/event_loop.c index 54b83db..1f1b789 100644 --- a/src/event_loop.c +++ b/src/event_loop.c @@ -18,80 +18,80 @@ SIMPLEQ_HEAD(event_fd_queue, event_fd); struct event_fd { - int fd; - short events; - event_handler handler; - void* arg; - int once; + int fd; + short events; + event_handler handler; + void* arg; + int once; - SIMPLEQ_ENTRY(event_fd) entries; + SIMPLEQ_ENTRY(event_fd) entries; }; static struct event_fd* event_fd_create(int fd, short events, event_handler handler, void* arg) { - struct event_fd* res = calloc(1, sizeof(struct event_fd)); - if (!res) { - log_errno("calloc"); - return NULL; - } + struct event_fd* res = calloc(1, sizeof(struct event_fd)); + if (!res) { + log_errno("calloc"); + return NULL; + } - res->fd = fd; - res->events = events; - res->handler = handler; - res->arg = arg; - res->once = 0; + res->fd = fd; + res->events = events; + res->handler = handler; + res->arg = arg; + res->once = 0; - return res; + return res; } static void event_fd_destroy(struct event_fd* entry) { - free(entry); + free(entry); } static void event_fd_queue_create(struct event_fd_queue* queue) { - SIMPLEQ_INIT(queue); + SIMPLEQ_INIT(queue); } static void event_fd_queue_destroy(struct event_fd_queue* queue) { - struct event_fd* entry1 = SIMPLEQ_FIRST(queue); - while (entry1) { - struct event_fd* entry2 = SIMPLEQ_NEXT(entry1, entries); - event_fd_destroy(entry1); - entry1 = entry2; - } - SIMPLEQ_INIT(queue); + struct event_fd* entry1 = SIMPLEQ_FIRST(queue); + while (entry1) { + struct event_fd* entry2 = SIMPLEQ_NEXT(entry1, entries); + event_fd_destroy(entry1); + entry1 = entry2; + } + SIMPLEQ_INIT(queue); } struct event_loop { - nfds_t nfds; - struct event_fd_queue entries; + nfds_t nfds; + struct event_fd_queue entries; }; int event_loop_create(struct event_loop** _loop) { - int ret = 0; + int ret = 0; - struct event_loop* loop = calloc(1, sizeof(struct event_loop)); - if (!loop) { - log_errno("calloc"); - return -1; - } - *_loop = loop; + struct event_loop* loop = calloc(1, sizeof(struct event_loop)); + if (!loop) { + log_errno("calloc"); + return -1; + } + *_loop = loop; - event_fd_queue_create(&loop->entries); + event_fd_queue_create(&loop->entries); - return ret; + return ret; } void event_loop_destroy(struct event_loop* loop) { - event_fd_queue_destroy(&loop->entries); - free(loop); + event_fd_queue_destroy(&loop->entries); + free(loop); } static void event_loop_add_internal(struct event_loop* loop, struct event_fd* entry) { - log_debug("Adding descriptor %d to event loop\n", entry->fd); + log_debug("Adding descriptor %d to event loop\n", entry->fd); - nfds_t nfds = loop->nfds + 1; - SIMPLEQ_INSERT_TAIL(&loop->entries, entry, entries); - loop->nfds = nfds; + nfds_t nfds = loop->nfds + 1; + SIMPLEQ_INSERT_TAIL(&loop->entries, entry, entries); + loop->nfds = nfds; } int event_loop_add(struct event_loop* loop, @@ -99,11 +99,11 @@ int event_loop_add(struct event_loop* loop, short events, event_handler handler, void* arg) { - struct event_fd* entry = event_fd_create(fd, events, handler, arg); - if (!entry) - return -1; - event_loop_add_internal(loop, entry); - return 0; + struct event_fd* entry = event_fd_create(fd, events, handler, arg); + if (!entry) + return -1; + event_loop_add_internal(loop, entry); + return 0; } int event_loop_add_once(struct event_loop* loop, @@ -111,117 +111,117 @@ int event_loop_add_once(struct event_loop* loop, short events, event_handler handler, void* arg) { - struct event_fd* entry = event_fd_create(fd, events, handler, arg); - if (!entry) - return -1; - entry->once = 1; - event_loop_add_internal(loop, entry); - return 0; + struct event_fd* entry = event_fd_create(fd, events, handler, arg); + if (!entry) + return -1; + entry->once = 1; + event_loop_add_internal(loop, entry); + return 0; } static void event_loop_remove(struct event_loop* loop, struct event_fd* entry) { - log_debug("Removing descriptor %d from event loop\n", entry->fd); + log_debug("Removing descriptor %d from event loop\n", entry->fd); - SIMPLEQ_REMOVE(&loop->entries, entry, event_fd, entries); - event_fd_destroy(entry); - --loop->nfds; + SIMPLEQ_REMOVE(&loop->entries, entry, event_fd, entries); + event_fd_destroy(entry); + --loop->nfds; } static char* append_event(char* buf, size_t sz, char* ptr, const char* event) { - if (ptr > buf) - ptr = string_append(ptr, buf + sz, ","); - return string_append(ptr, buf + sz, event); + if (ptr > buf) + ptr = string_append(ptr, buf + sz, ","); + return string_append(ptr, buf + sz, event); } static char* events_to_string(short events) { - const size_t sz = 128; - char* buf = calloc(1, sz); - if (!buf) - return NULL; + const size_t sz = 128; + char* buf = calloc(1, sz); + if (!buf) + return NULL; - char* ptr = buf; + char* ptr = buf; - if (events & POLLNVAL) - ptr = append_event(buf, sz, ptr, "POLLNVAL"); - if (events & POLLERR) - ptr = append_event(buf, sz, ptr, "POLLERR"); - if (events & POLLHUP) - ptr = append_event(buf, sz, ptr, "POLLHUP"); - if (events & POLLRDHUP) - ptr = append_event(buf, sz, ptr, "POLLRDHUP"); - if (events & POLLPRI) - ptr = append_event(buf, sz, ptr, "POLLPRI"); - if (events & POLLIN) - ptr = append_event(buf, sz, ptr, "POLLIN"); - if (events & POLLOUT) - ptr = append_event(buf, sz, ptr, "POLLOUT"); + if (events & POLLNVAL) + ptr = append_event(buf, sz, ptr, "POLLNVAL"); + if (events & POLLERR) + ptr = append_event(buf, sz, ptr, "POLLERR"); + if (events & POLLHUP) + ptr = append_event(buf, sz, ptr, "POLLHUP"); + if (events & POLLRDHUP) + ptr = append_event(buf, sz, ptr, "POLLRDHUP"); + if (events & POLLPRI) + ptr = append_event(buf, sz, ptr, "POLLPRI"); + if (events & POLLIN) + ptr = append_event(buf, sz, ptr, "POLLIN"); + if (events & POLLOUT) + ptr = append_event(buf, sz, ptr, "POLLOUT"); - return buf; + return buf; } static struct pollfd* make_pollfds(const struct event_loop* loop) { - struct pollfd* fds = calloc(loop->nfds, sizeof(struct pollfd)); - if (!fds) { - log_errno("calloc"); - return NULL; - } + struct pollfd* fds = calloc(loop->nfds, sizeof(struct pollfd)); + if (!fds) { + log_errno("calloc"); + return NULL; + } - struct event_fd* entry = SIMPLEQ_FIRST(&loop->entries); - for (nfds_t i = 0; i < loop->nfds; ++i, entry = SIMPLEQ_NEXT(entry, entries)) { - fds[i].fd = entry->fd; - fds[i].events = entry->events; - } + struct event_fd* entry = SIMPLEQ_FIRST(&loop->entries); + for (nfds_t i = 0; i < loop->nfds; ++i, entry = SIMPLEQ_NEXT(entry, entries)) { + fds[i].fd = entry->fd; + fds[i].events = entry->events; + } - log_debug("Descriptors:\n"); - for (nfds_t i = 0; i < loop->nfds; ++i) { - char* events = events_to_string(fds[i].events); - log_debug(" %d (%s)\n", fds[i].fd, events ? events : ""); - free(events); - } + log_debug("Descriptors:\n"); + for (nfds_t i = 0; i < loop->nfds; ++i) { + char* events = events_to_string(fds[i].events); + log_debug(" %d (%s)\n", fds[i].fd, events ? events : ""); + free(events); + } - return fds; + return fds; } int event_loop_run(struct event_loop* loop) { - /* Cache the number of event descriptors so that event handlers can - * append new ones. */ - const nfds_t nfds = loop->nfds; + /* Cache the number of event descriptors so that event handlers can + * append new ones. */ + const nfds_t nfds = loop->nfds; - struct pollfd* fds = make_pollfds(loop); - if (!fds) - return -1; + struct pollfd* fds = make_pollfds(loop); + if (!fds) + return -1; - int ret = poll(fds, nfds, -1); - if (ret < 0) { - log_errno("poll"); - return ret; - } - ret = 0; + int ret = poll(fds, nfds, -1); + if (ret < 0) { + log_errno("poll"); + return ret; + } + ret = 0; - struct event_fd* entry = SIMPLEQ_FIRST(&loop->entries); - for (nfds_t i = 0; i < nfds; ++i) { - struct event_fd* next = SIMPLEQ_NEXT(entry, entries); + struct event_fd* entry = SIMPLEQ_FIRST(&loop->entries); + for (nfds_t i = 0; i < nfds; ++i) { + struct event_fd* next = SIMPLEQ_NEXT(entry, entries); - if (!fds[i].revents) - goto next; + if (!fds[i].revents) + goto next; - char* events = events_to_string(fds[i].revents); - log_debug("Descriptor %d is ready: %s\n", fds[i].fd, events ? events : ""); - free(events); + char* events = events_to_string(fds[i].revents); + log_debug("Descriptor %d is ready: %s\n", fds[i].fd, events ? events : ""); + free(events); - /* Execute all handlers but notice if any of them fail. */ - const int handler_ret = entry->handler(loop, fds[i].fd, fds[i].revents, entry->arg); - if (handler_ret < 0) - ret = handler_ret; + /* Execute all handlers but notice if any of them fail. */ + const int handler_ret = entry->handler(loop, fds[i].fd, fds[i].revents, entry->arg); + if (handler_ret < 0) + ret = handler_ret; - if (entry->once) - event_loop_remove(loop, entry); + if (entry->once) + event_loop_remove(loop, entry); - next: - entry = next; - continue; - } + next: + entry = next; + continue; + } - free(fds); - return ret; + free(fds); + return ret; } @@ -21,137 +21,137 @@ static int unlink_cb(const char* fpath, UNUSED const struct stat* sb, UNUSED int typeflag, UNUSED struct FTW* ftwbuf) { - int ret = 0; + int ret = 0; - ret = remove(fpath); - if (ret < 0) { - log_errno("remove"); - return ret; - } + ret = remove(fpath); + if (ret < 0) { + log_errno("remove"); + return ret; + } - return ret; + return ret; } int rm_rf(const char* dir) { - log("Recursively removing directory: %s\n", dir); - return nftw(dir, unlink_cb, 64, FTW_DEPTH | FTW_PHYS); + log("Recursively removing directory: %s\n", dir); + return nftw(dir, unlink_cb, 64, FTW_DEPTH | FTW_PHYS); } int chdir_wrapper(const char* dir, char** old) { - int ret = 0; + int ret = 0; - if (old) { - *old = get_current_dir_name(); - if (!*old) { - log_errno("get_current_dir_name"); - return -1; - } - } + if (old) { + *old = get_current_dir_name(); + if (!*old) { + log_errno("get_current_dir_name"); + return -1; + } + } - ret = chdir(dir); - if (ret < 0) { - log_errno("chdir"); - goto free_old; - } + ret = chdir(dir); + if (ret < 0) { + log_errno("chdir"); + goto free_old; + } - return ret; + return ret; free_old: - if (old) - free(*old); + if (old) + free(*old); - return ret; + return ret; } char* readlink_wrapper(const char* path) { - size_t current_size = 256; - char* buf = NULL; + size_t current_size = 256; + char* buf = NULL; - while (1) { - char* tmp_buf = realloc(buf, current_size); - if (!tmp_buf) { - log_errno("realloc"); - goto free; - } - buf = tmp_buf; + while (1) { + char* tmp_buf = realloc(buf, current_size); + if (!tmp_buf) { + log_errno("realloc"); + goto free; + } + buf = tmp_buf; - ssize_t res = readlink(path, buf, current_size); - if (res < 0) { - log_errno("readlink"); - goto free; - } + ssize_t res = readlink(path, buf, current_size); + if (res < 0) { + log_errno("readlink"); + goto free; + } - if ((size_t)res == current_size) { - current_size *= 2; - continue; - } + if ((size_t)res == current_size) { + current_size *= 2; + continue; + } - buf[res] = '\0'; - break; - } + buf[res] = '\0'; + break; + } - return buf; + return buf; free: - free(buf); + free(buf); - return NULL; + return NULL; } int file_dup(int fd) { - int ret = 0; + int ret = 0; - ret = fcntl(fd, F_DUPFD_CLOEXEC, 0); - if (ret < 0) { - log_errno("fcntl"); - return ret; - } + ret = fcntl(fd, F_DUPFD_CLOEXEC, 0); + if (ret < 0) { + log_errno("fcntl"); + return ret; + } - return ret; + return ret; } void file_close(int fd) { - log_errno_if(close(fd), "close"); + log_errno_if(close(fd), "close"); } int file_exists(const char* path) { - struct stat stat; - int ret = lstat(path, &stat); - return !ret && S_ISREG(stat.st_mode); + struct stat stat; + int ret = lstat(path, &stat); + return !ret && S_ISREG(stat.st_mode); } int file_read(int fd, unsigned char** _contents, size_t* _size) { - size_t alloc_size = 256; - unsigned char* contents = NULL; - size_t size = 0; + size_t alloc_size = 256; + unsigned char* contents = NULL; + size_t size = 0; - while (1) { - unsigned char* tmp_contents = realloc(contents, alloc_size); - if (!tmp_contents) { - log_errno("realloc"); - free(contents); - return -1; - } - contents = tmp_contents; + while (1) { + unsigned char* tmp_contents = realloc(contents, alloc_size); + if (!tmp_contents) { + log_errno("realloc"); + free(contents); + return -1; + } + contents = tmp_contents; - ssize_t read_size = read(fd, contents + size, alloc_size - size); + ssize_t read_size = read(fd, contents + size, alloc_size - size); - if (read_size < 0) { - log_errno("read"); - free(contents); - return read_size; - } + if (read_size < 0) { + log_errno("read"); + free(contents); + return read_size; + } - if (!read_size) { - *_contents = contents; - *_size = size; - return 0; - } + if (!read_size) { + *_contents = contents; + *_size = size; + return 0; + } - size += read_size; + size += read_size; - if (size == alloc_size) { - alloc_size *= 2; - } - } + if (size == alloc_size) { + alloc_size *= 2; + } + } } @@ -13,100 +13,100 @@ #include <stdlib.h> -#define git_log_err(fn) \ - do { \ - const git_error* error = git_error_last(); \ - const char* msg = error && error->message ? error->message : "???"; \ - log_err("%s: %s\n", fn, msg); \ - } while (0) +#define git_log_err(fn) \ + do { \ + const git_error* error = git_error_last(); \ + const char* msg = error && error->message ? error->message : "???"; \ + log_err("%s: %s\n", fn, msg); \ + } while (0) int libgit_init(void) { - int ret = 0; + int ret = 0; - ret = git_libgit2_init(); - if (ret < 0) { - git_log_err("git_libgit2_init"); - return ret; - } + ret = git_libgit2_init(); + if (ret < 0) { + git_log_err("git_libgit2_init"); + return ret; + } - return 0; + return 0; } void libgit_shutdown(void) { - git_libgit2_shutdown(); + git_libgit2_shutdown(); } int libgit_clone(git_repository** repo, const char* url, const char* dir) { - git_clone_options opts; - int ret = 0; + git_clone_options opts; + int ret = 0; - log("Cloning git repository from %s to %s\n", url, dir); + log("Cloning git repository from %s to %s\n", url, dir); - ret = git_clone_options_init(&opts, GIT_CLONE_OPTIONS_VERSION); - if (ret < 0) { - git_log_err("git_clone_options_init"); - return ret; - } - opts.checkout_opts.checkout_strategy = GIT_CHECKOUT_NONE; + ret = git_clone_options_init(&opts, GIT_CLONE_OPTIONS_VERSION); + if (ret < 0) { + git_log_err("git_clone_options_init"); + return ret; + } + opts.checkout_opts.checkout_strategy = GIT_CHECKOUT_NONE; - ret = git_clone(repo, url, dir, &opts); - if (ret < 0) { - git_log_err("git_clone"); - return ret; - } + ret = git_clone(repo, url, dir, &opts); + if (ret < 0) { + git_log_err("git_clone"); + return ret; + } - return 0; + return 0; } int libgit_clone_to_tmp(git_repository** repo, const char* url) { - char dir[] = "/tmp/git.XXXXXX"; + char dir[] = "/tmp/git.XXXXXX"; - if (!mkdtemp(dir)) { - log_errno("mkdtemp"); - return -1; - } + if (!mkdtemp(dir)) { + log_errno("mkdtemp"); + return -1; + } - return libgit_clone(repo, url, dir); + return libgit_clone(repo, url, dir); } void libgit_repository_free(git_repository* repo) { - git_repository_free(repo); + git_repository_free(repo); } int libgit_checkout(git_repository* repo, const char* rev) { - git_checkout_options opts; - git_object* obj; - int ret = 0; + git_checkout_options opts; + git_object* obj; + int ret = 0; - log("Checking out revision %s\n", rev); + log("Checking out revision %s\n", rev); - ret = git_revparse_single(&obj, repo, rev); - if (ret < 0) { - git_log_err("git_revparse_single"); - return ret; - } + ret = git_revparse_single(&obj, repo, rev); + if (ret < 0) { + git_log_err("git_revparse_single"); + return ret; + } - ret = git_checkout_options_init(&opts, GIT_CHECKOUT_OPTIONS_VERSION); - if (ret < 0) { - git_log_err("git_checkout_options_init"); - goto free_obj; - } - opts.checkout_strategy = GIT_CHECKOUT_FORCE; + ret = git_checkout_options_init(&opts, GIT_CHECKOUT_OPTIONS_VERSION); + if (ret < 0) { + git_log_err("git_checkout_options_init"); + goto free_obj; + } + opts.checkout_strategy = GIT_CHECKOUT_FORCE; - ret = git_checkout_tree(repo, obj, &opts); - if (ret < 0) { - git_log_err("git_checkout_tree"); - goto free_obj; - } + ret = git_checkout_tree(repo, obj, &opts); + if (ret < 0) { + git_log_err("git_checkout_tree"); + goto free_obj; + } - ret = git_repository_set_head_detached(repo, git_object_id(obj)); - if (ret < 0) { - git_log_err("git_repository_set_head_detached"); - goto free_obj; - } + ret = git_repository_set_head_detached(repo, git_object_id(obj)); + if (ret < 0) { + git_log_err("git_repository_set_head_detached"); + goto free_obj; + } free_obj: - git_object_free(obj); + git_object_free(obj); - return ret; + return ret; } @@ -19,240 +19,240 @@ #include <stdlib.h> #include <string.h> -#define libjson_errno(fn) \ - do { \ - log_err("JSON: %s failed\n", fn); \ - } while (0) +#define libjson_errno(fn) \ + do { \ + log_err("JSON: %s failed\n", fn); \ + } while (0) void libjson_free(struct json_object* obj) { - json_object_put(obj); + json_object_put(obj); } static const char* libjson_to_string_internal(struct json_object* obj, int flags) { - const char* result = json_object_to_json_string_ext(obj, flags); - if (!result) { - libjson_errno("json_object_to_json_string"); - return NULL; - } - return result; + const char* result = json_object_to_json_string_ext(obj, flags); + if (!result) { + libjson_errno("json_object_to_json_string"); + return NULL; + } + return result; } const char* libjson_to_string(struct json_object* obj) { - return libjson_to_string_internal(obj, JSON_C_TO_STRING_SPACED); + return libjson_to_string_internal(obj, JSON_C_TO_STRING_SPACED); } const char* libjson_to_string_pretty(struct json_object* obj) { - return libjson_to_string_internal(obj, JSON_C_TO_STRING_SPACED | JSON_C_TO_STRING_PRETTY); + return libjson_to_string_internal(obj, JSON_C_TO_STRING_SPACED | JSON_C_TO_STRING_PRETTY); } struct json_object* libjson_from_string(const char* src) { - enum json_tokener_error error; + enum json_tokener_error error; - struct json_object* result = json_tokener_parse_verbose(src, &error); - if (!result) { - libjson_errno("json_tokener_parse_verbose"); - log_err("JSON: parsing failed: %s\n", json_tokener_error_desc(error)); - return NULL; - } + struct json_object* result = json_tokener_parse_verbose(src, &error); + if (!result) { + libjson_errno("json_tokener_parse_verbose"); + log_err("JSON: parsing failed: %s\n", json_tokener_error_desc(error)); + return NULL; + } - return result; + return result; } int libjson_clone(const struct json_object* obj, const char* key, struct json_object** _value) { - int ret = 0; + int ret = 0; - struct json_object* old_value = NULL; - ret = libjson_get(obj, key, &old_value); - if (ret < 0) - return ret; + struct json_object* old_value = NULL; + ret = libjson_get(obj, key, &old_value); + if (ret < 0) + return ret; - struct json_object* new_value = NULL; - ret = json_object_deep_copy(old_value, &new_value, NULL); - if (ret < 0) - return ret; + struct json_object* new_value = NULL; + ret = json_object_deep_copy(old_value, &new_value, NULL); + if (ret < 0) + return ret; - *_value = new_value; - return ret; + *_value = new_value; + return ret; } int libjson_send(struct json_object* obj, int fd) { - int ret = 0; + int ret = 0; - const char* str = libjson_to_string(obj); - if (!str) - return -1; + const char* str = libjson_to_string(obj); + if (!str) + return -1; - struct buf* buf = NULL; - ret = buf_create_from_string(&buf, str); - if (ret < 0) - return ret; + struct buf* buf = NULL; + ret = buf_create_from_string(&buf, str); + if (ret < 0) + return ret; - ret = net_send_buf(fd, buf); - buf_destroy(buf); - return ret; + ret = net_send_buf(fd, buf); + buf_destroy(buf); + return ret; } struct json_object* libjson_recv(int fd) { - struct json_object* result = NULL; - int ret = 0; + struct json_object* result = NULL; + int ret = 0; - struct buf* buf = NULL; - ret = net_recv_buf(fd, &buf); - if (ret < 0) - return NULL; + struct buf* buf = NULL; + ret = net_recv_buf(fd, &buf); + if (ret < 0) + return NULL; - result = libjson_from_string((const char*)buf_get_data(buf)); - if (!result) - goto destroy_buf; + result = libjson_from_string((const char*)buf_get_data(buf)); + if (!result) + goto destroy_buf; destroy_buf: - free((void*)buf_get_data(buf)); - buf_destroy(buf); + free((void*)buf_get_data(buf)); + buf_destroy(buf); - return result; + return result; } int libjson_new_object(struct json_object** _obj) { - struct json_object* obj = json_object_new_object(); - if (!obj) { - libjson_errno("json_object_new_object"); - return -1; - } - *_obj = obj; - return 0; + struct json_object* obj = json_object_new_object(); + if (!obj) { + libjson_errno("json_object_new_object"); + return -1; + } + *_obj = obj; + return 0; } int libjson_new_array(struct json_object** _arr) { - struct json_object* arr = json_object_new_array(); - if (!arr) { - libjson_errno("json_object_new_array"); - return -1; - } - *_arr = arr; - return 0; + struct json_object* arr = json_object_new_array(); + if (!arr) { + libjson_errno("json_object_new_array"); + return -1; + } + *_arr = arr; + return 0; } int libjson_has(const struct json_object* obj, const char* key) { - return json_object_object_get_ex(obj, key, NULL); + return json_object_object_get_ex(obj, key, NULL); } int libjson_get(const struct json_object* obj, const char* key, struct json_object** value) { - if (!libjson_has(obj, key)) { - log_err("JSON: key is missing: %s\n", key); - return -1; - } + if (!libjson_has(obj, key)) { + log_err("JSON: key is missing: %s\n", key); + return -1; + } - return json_object_object_get_ex(obj, key, value); + return json_object_object_get_ex(obj, key, value); } int libjson_get_string(const struct json_object* obj, const char* key, const char** _value) { - struct json_object* value = NULL; + struct json_object* value = NULL; - int ret = libjson_get(obj, key, &value); - if (ret < 0) - return ret; + int ret = libjson_get(obj, key, &value); + if (ret < 0) + return ret; - if (!json_object_is_type(value, json_type_string)) { - log_err("JSON: key is not a string: %s\n", key); - return -1; - } + if (!json_object_is_type(value, json_type_string)) { + log_err("JSON: key is not a string: %s\n", key); + return -1; + } - *_value = json_object_get_string(value); - return 0; + *_value = json_object_get_string(value); + return 0; } int libjson_get_int(const struct json_object* obj, const char* key, int64_t* _value) { - struct json_object* value = NULL; + struct json_object* value = NULL; - int ret = libjson_get(obj, key, &value); - if (ret < 0) - return ret; + int ret = libjson_get(obj, key, &value); + if (ret < 0) + return ret; - if (!json_object_is_type(value, json_type_int)) { - log_err("JSON: key is not an integer: %s\n", key); - return -1; - } + if (!json_object_is_type(value, json_type_int)) { + log_err("JSON: key is not an integer: %s\n", key); + return -1; + } - errno = 0; - int64_t tmp = json_object_get_int64(value); - if (errno) { - log_err("JSON: failed to parse integer from key: %s\n", key); - return -1; - } + errno = 0; + int64_t tmp = json_object_get_int64(value); + if (errno) { + log_err("JSON: failed to parse integer from key: %s\n", key); + return -1; + } - *_value = tmp; - return 0; + *_value = tmp; + return 0; } static int libjson_set_internal(struct json_object* obj, const char* key, struct json_object* value, unsigned flags) { - int ret = 0; + int ret = 0; - ret = json_object_object_add_ex(obj, key, value, flags); - if (ret < 0) { - libjson_errno("json_object_object_add_ex"); - return ret; - } + ret = json_object_object_add_ex(obj, key, value, flags); + if (ret < 0) { + libjson_errno("json_object_object_add_ex"); + return ret; + } - return 0; + return 0; } static int libjson_set_string_internal(struct json_object* obj, const char* key, const char* _value, unsigned flags) { - struct json_object* value = json_object_new_string(_value); - if (!value) { - libjson_errno("json_object_new_string"); - return -1; - } + struct json_object* value = json_object_new_string(_value); + if (!value) { + libjson_errno("json_object_new_string"); + return -1; + } - int ret = libjson_set_internal(obj, key, value, flags); - if (ret < 0) - goto free_value; + int ret = libjson_set_internal(obj, key, value, flags); + if (ret < 0) + goto free_value; - return ret; + return ret; free_value: - json_object_put(value); + json_object_put(value); - return ret; + return ret; } static int libjson_set_int_internal(struct json_object* obj, const char* key, int64_t _value, unsigned flags) { - struct json_object* value = json_object_new_int64(_value); - if (!value) { - libjson_errno("json_object_new_int"); - return -1; - } + struct json_object* value = json_object_new_int64(_value); + if (!value) { + libjson_errno("json_object_new_int"); + return -1; + } - int ret = libjson_set_internal(obj, key, value, flags); - if (ret < 0) - goto free_value; + int ret = libjson_set_internal(obj, key, value, flags); + if (ret < 0) + goto free_value; - return ret; + return ret; free_value: - json_object_put(value); + json_object_put(value); - return ret; + return ret; } int libjson_set(struct json_object* obj, const char* key, struct json_object* value) { - return libjson_set_internal(obj, key, value, 0); + return libjson_set_internal(obj, key, value, 0); } int libjson_set_string(struct json_object* obj, const char* key, const char* value) { - return libjson_set_string_internal(obj, key, value, 0); + return libjson_set_string_internal(obj, key, value, 0); } int libjson_set_int(struct json_object* obj, const char* key, int64_t value) { - return libjson_set_int_internal(obj, key, value, 0); + return libjson_set_int_internal(obj, key, value, 0); } #ifndef JSON_C_OBJECT_ADD_CONSTANT_KEY @@ -261,22 +261,22 @@ int libjson_set_int(struct json_object* obj, const char* key, int64_t value) { static const unsigned json_const_key_flags = JSON_C_OBJECT_ADD_CONSTANT_KEY; int libjson_set_const_key(struct json_object* obj, const char* key, struct json_object* value) { - return libjson_set_internal(obj, key, value, json_const_key_flags); + return libjson_set_internal(obj, key, value, json_const_key_flags); } int libjson_set_string_const_key(struct json_object* obj, const char* key, const char* value) { - return libjson_set_string_internal(obj, key, value, json_const_key_flags); + return libjson_set_string_internal(obj, key, value, json_const_key_flags); } int libjson_set_int_const_key(struct json_object* obj, const char* key, int64_t value) { - return libjson_set_int_internal(obj, key, value, json_const_key_flags); + return libjson_set_int_internal(obj, key, value, json_const_key_flags); } int libjson_append(struct json_object* arr, struct json_object* elem) { - int ret = json_object_array_add(arr, elem); - if (ret < 0) { - libjson_errno("json_object_array_add"); - return ret; - } - return ret; + int ret = json_object_array_add(arr, elem); + if (ret < 0) { + libjson_errno("json_object_array_add"); + return ret; + } + return ret; } diff --git a/src/json_rpc.c b/src/json_rpc.c index 2ab4624..6bf9fc5 100644 --- a/src/json_rpc.c +++ b/src/json_rpc.c @@ -18,11 +18,11 @@ #include <string.h> struct jsonrpc_request { - struct json_object* impl; + struct json_object* impl; }; struct jsonrpc_response { - struct json_object* impl; + struct json_object* impl; }; static const char* const jsonrpc_key_version = "jsonrpc"; @@ -33,103 +33,101 @@ static const char* const jsonrpc_key_params = "params"; static const char* const jsonrpc_value_version = "2.0"; static int jsonrpc_check_version(struct json_object* obj) { - const char* key = jsonrpc_key_version; - const char* version = NULL; + const char* key = jsonrpc_key_version; + const char* version = NULL; - int ret = libjson_get_string(obj, key, &version); - if (ret < 0) - return ret; + int ret = libjson_get_string(obj, key, &version); + if (ret < 0) + return ret; - if (strcmp(version, jsonrpc_value_version)) { - log_err("JSON-RPC: invalid '%s' value: %s\n", key, version); - return -1; - } + if (strcmp(version, jsonrpc_value_version)) { + log_err("JSON-RPC: invalid '%s' value: %s\n", key, version); + return -1; + } - return 0; + return 0; } static int jsonrpc_set_version(struct json_object* obj) { - return libjson_set_string_const_key(obj, jsonrpc_key_version, jsonrpc_value_version); + return libjson_set_string_const_key(obj, jsonrpc_key_version, jsonrpc_value_version); } static _Atomic int jsonrpc_id_counter = 1; int jsonrpc_generate_request_id(void) { - return jsonrpc_id_counter++; + return jsonrpc_id_counter++; } static int jsonrpc_check_id_type(struct json_object* id) { - if (!json_object_is_type(id, json_type_string) && !json_object_is_type(id, json_type_int)) { - log_err("JSON-RPC: key '%s' must be either an integer or a string\n", - jsonrpc_key_id); - return -1; - } - return 0; + if (!json_object_is_type(id, json_type_string) && !json_object_is_type(id, json_type_int)) { + log_err("JSON-RPC: key '%s' must be either an integer or a string\n", jsonrpc_key_id); + return -1; + } + return 0; } static int jsonrpc_check_id(struct json_object* obj, int required) { - const char* key = jsonrpc_key_id; + const char* key = jsonrpc_key_id; - if (!libjson_has(obj, key)) { - if (!required) - return 0; - log_err("JSON-RPC: key is missing: %s\n", key); - return -1; - } + if (!libjson_has(obj, key)) { + if (!required) + return 0; + log_err("JSON-RPC: key is missing: %s\n", key); + return -1; + } - struct json_object* id = NULL; + struct json_object* id = NULL; - int ret = libjson_get(obj, key, &id); - if (ret < 0) - return ret; - return jsonrpc_check_id_type(id); + int ret = libjson_get(obj, key, &id); + if (ret < 0) + return ret; + return jsonrpc_check_id_type(id); } static int jsonrpc_set_id(struct json_object* obj, int id) { - return libjson_set_int_const_key(obj, jsonrpc_key_id, id); + return libjson_set_int_const_key(obj, jsonrpc_key_id, id); } static int jsonrpc_check_method(struct json_object* obj) { - const char* key = jsonrpc_key_method; - const char* method = NULL; - return libjson_get_string(obj, key, &method); + const char* key = jsonrpc_key_method; + const char* method = NULL; + return libjson_get_string(obj, key, &method); } static int jsonrpc_set_method(struct json_object* obj, const char* method) { - return libjson_set_string_const_key(obj, jsonrpc_key_method, method); + return libjson_set_string_const_key(obj, jsonrpc_key_method, method); } static int jsonrpc_check_params_type(struct json_object* params) { - if (!json_object_is_type(params, json_type_object) && - !json_object_is_type(params, json_type_array)) { - log_err("JSON-RPC: key '%s' must be either an object or an array\n", - jsonrpc_key_params); - return -1; - } - return 0; + if (!json_object_is_type(params, json_type_object) && + !json_object_is_type(params, json_type_array)) { + log_err("JSON-RPC: key '%s' must be either an object or an array\n", jsonrpc_key_params); + return -1; + } + return 0; } static int jsonrpc_check_params(struct json_object* obj) { - const char* key = jsonrpc_key_params; + const char* key = jsonrpc_key_params; - if (!libjson_has(obj, key)) - return 0; + if (!libjson_has(obj, key)) + return 0; - struct json_object* params = NULL; + struct json_object* params = NULL; - int ret = libjson_get(obj, key, ¶ms); - if (ret < 0) - return ret; - return jsonrpc_check_params_type(params); + int ret = libjson_get(obj, key, ¶ms); + if (ret < 0) + return ret; + return jsonrpc_check_params_type(params); } static int jsonrpc_set_params(struct json_object* obj, struct json_object* params) { - const char* key = jsonrpc_key_params; + const char* key = jsonrpc_key_params; - int ret = jsonrpc_check_params_type(params); - if (ret < 0) - return ret; - return libjson_set_const_key(obj, key, params); + int ret = jsonrpc_check_params_type(params); + if (ret < 0) + return ret; + return libjson_set_const_key(obj, key, params); } static const char* const jsonrpc_key_result = "result"; @@ -139,387 +137,387 @@ static const char* const jsonrpc_key_code = "code"; static const char* const jsonrpc_key_message = "message"; static int jsonrpc_check_error(struct json_object* obj) { - const char* key = jsonrpc_key_error; - struct json_object* error = NULL; + const char* key = jsonrpc_key_error; + struct json_object* error = NULL; - int ret = libjson_get(obj, key, &error); - if (ret < 0) - return ret; + int ret = libjson_get(obj, key, &error); + if (ret < 0) + return ret; - int64_t code = -1; + int64_t code = -1; - ret = libjson_get_int(error, jsonrpc_key_code, &code); - if (ret < 0) { - log_err("JSON-RPC: key is missing or not an integer: %s\n", jsonrpc_key_code); - return -1; - } + ret = libjson_get_int(error, jsonrpc_key_code, &code); + if (ret < 0) { + log_err("JSON-RPC: key is missing or not an integer: %s\n", jsonrpc_key_code); + return -1; + } - const char* message = NULL; + const char* message = NULL; - ret = libjson_get_string(error, jsonrpc_key_message, &message); - if (ret < 0) { - log_err("JSON-RPC: key is missing or not a string: %s\n", jsonrpc_key_message); - return -1; - } + ret = libjson_get_string(error, jsonrpc_key_message, &message); + if (ret < 0) { + log_err("JSON-RPC: key is missing or not a string: %s\n", jsonrpc_key_message); + return -1; + } - return ret; + return ret; } static int jsonrpc_check_result_or_error(struct json_object* obj) { - const char* key_result = jsonrpc_key_result; - const char* key_error = jsonrpc_key_error; + const char* key_result = jsonrpc_key_result; + const char* key_error = jsonrpc_key_error; - if (!libjson_has(obj, key_result) && !libjson_has(obj, key_error)) { - log_err("JSON-RPC: either '%s' or '%s' must be present\n", key_result, key_error); - return -1; - } + if (!libjson_has(obj, key_result) && !libjson_has(obj, key_error)) { + log_err("JSON-RPC: either '%s' or '%s' must be present\n", key_result, key_error); + return -1; + } - if (libjson_has(obj, key_result)) - return 0; + if (libjson_has(obj, key_result)) + return 0; - return jsonrpc_check_error(obj); + return jsonrpc_check_error(obj); } static int jsonrpc_request_create_internal(struct jsonrpc_request** _request, int* id, const char* method, struct json_object* params) { - int ret = 0; + int ret = 0; - struct jsonrpc_request* request = malloc(sizeof(struct jsonrpc_request)); - if (!request) { - log_errno("malloc"); - ret = -1; - goto exit; - } + struct jsonrpc_request* request = malloc(sizeof(struct jsonrpc_request)); + if (!request) { + log_errno("malloc"); + ret = -1; + goto exit; + } - ret = libjson_new_object(&request->impl); - if (ret < 0) - goto free; + ret = libjson_new_object(&request->impl); + if (ret < 0) + goto free; - ret = jsonrpc_set_version(request->impl); - if (ret < 0) - goto free_impl; + ret = jsonrpc_set_version(request->impl); + if (ret < 0) + goto free_impl; - if (id) { - ret = jsonrpc_set_id(request->impl, *id); - if (ret < 0) - goto free_impl; - } + if (id) { + ret = jsonrpc_set_id(request->impl, *id); + if (ret < 0) + goto free_impl; + } - ret = jsonrpc_set_method(request->impl, method); - if (ret < 0) - goto free_impl; + ret = jsonrpc_set_method(request->impl, method); + if (ret < 0) + goto free_impl; - if (params) { - ret = jsonrpc_set_params(request->impl, params); - if (ret < 0) - goto free_impl; - } + if (params) { + ret = jsonrpc_set_params(request->impl, params); + if (ret < 0) + goto free_impl; + } - *_request = request; - goto exit; + *_request = request; + goto exit; free_impl: - libjson_free(request->impl); + libjson_free(request->impl); free: - free(request); + free(request); exit: - return ret; + return ret; } int jsonrpc_request_create(struct jsonrpc_request** _request, int id, const char* method, struct json_object* params) { - return jsonrpc_request_create_internal(_request, &id, method, params); + return jsonrpc_request_create_internal(_request, &id, method, params); } void jsonrpc_request_destroy(struct jsonrpc_request* request) { - libjson_free(request->impl); - free(request); + libjson_free(request->impl); + free(request); } int jsonrpc_notification_create(struct jsonrpc_request** _request, const char* method, struct json_object* params) { - return jsonrpc_request_create_internal(_request, NULL, method, params); + return jsonrpc_request_create_internal(_request, NULL, method, params); } int jsonrpc_request_is_notification(const struct jsonrpc_request* request) { - return !libjson_has(request->impl, jsonrpc_key_id); + return !libjson_has(request->impl, jsonrpc_key_id); } static int jsonrpc_request_from_json(struct jsonrpc_request** _request, struct json_object* impl) { - int ret = 0; + int ret = 0; - ret = jsonrpc_check_version(impl); - if (ret < 0) - return ret; - ret = jsonrpc_check_id(impl, 0); - if (ret < 0) - return ret; - ret = jsonrpc_check_method(impl); - if (ret < 0) - return ret; - ret = jsonrpc_check_params(impl); - if (ret < 0) - return ret; + ret = jsonrpc_check_version(impl); + if (ret < 0) + return ret; + ret = jsonrpc_check_id(impl, 0); + if (ret < 0) + return ret; + ret = jsonrpc_check_method(impl); + if (ret < 0) + return ret; + ret = jsonrpc_check_params(impl); + if (ret < 0) + return ret; - struct jsonrpc_request* request = malloc(sizeof(struct jsonrpc_request)); - if (!request) { - log_errno("malloc"); - return -1; - } - request->impl = impl; + struct jsonrpc_request* request = malloc(sizeof(struct jsonrpc_request)); + if (!request) { + log_errno("malloc"); + return -1; + } + request->impl = impl; - *_request = request; - return ret; + *_request = request; + return ret; } int jsonrpc_request_send(const struct jsonrpc_request* request, int fd) { - return libjson_send(request->impl, fd); + return libjson_send(request->impl, fd); } int jsonrpc_request_recv(struct jsonrpc_request** request, int fd) { - struct json_object* impl = libjson_recv(fd); - if (!impl) { - log_err("JSON-RPC: failed to receive request\n"); - return -1; - } + struct json_object* impl = libjson_recv(fd); + if (!impl) { + log_err("JSON-RPC: failed to receive request\n"); + return -1; + } - int ret = jsonrpc_request_from_json(request, impl); - if (ret < 0) - goto free_impl; + int ret = jsonrpc_request_from_json(request, impl); + if (ret < 0) + goto free_impl; - return ret; + return ret; free_impl: - libjson_free(impl); + libjson_free(impl); - return ret; + return ret; } const char* jsonrpc_request_get_method(const struct jsonrpc_request* request) { - const char* method = NULL; - int ret = libjson_get_string(request->impl, jsonrpc_key_method, &method); - if (ret < 0) { - /* Should never happen. */ - return NULL; - } - return method; + const char* method = NULL; + int ret = libjson_get_string(request->impl, jsonrpc_key_method, &method); + if (ret < 0) { + /* Should never happen. */ + return NULL; + } + return method; } static struct json_object* jsonrpc_request_create_params(struct jsonrpc_request* request) { - int ret = 0; - const char* const key = jsonrpc_key_params; + int ret = 0; + const char* const key = jsonrpc_key_params; - if (!libjson_has(request->impl, key)) { - struct json_object* params = NULL; + if (!libjson_has(request->impl, key)) { + struct json_object* params = NULL; - ret = libjson_new_object(¶ms); - if (ret < 0) - return NULL; + ret = libjson_new_object(¶ms); + if (ret < 0) + return NULL; - ret = libjson_set(request->impl, key, params); - if (ret < 0) { - libjson_free(params); - return NULL; - } - return params; - } + ret = libjson_set(request->impl, key, params); + if (ret < 0) { + libjson_free(params); + return NULL; + } + return params; + } - struct json_object* params = NULL; - ret = libjson_get(request->impl, key, ¶ms); - if (ret < 0) - return NULL; - return params; + struct json_object* params = NULL; + ret = libjson_get(request->impl, key, ¶ms); + if (ret < 0) + return NULL; + return params; } int jsonrpc_request_get_param_string(const struct jsonrpc_request* request, const char* name, const char** value) { - struct json_object* params = NULL; - int ret = libjson_get(request->impl, jsonrpc_key_params, ¶ms); - if (ret < 0) - return ret; - return libjson_get_string(params, name, value); + struct json_object* params = NULL; + int ret = libjson_get(request->impl, jsonrpc_key_params, ¶ms); + if (ret < 0) + return ret; + return libjson_get_string(params, name, value); } int jsonrpc_request_set_param_string(struct jsonrpc_request* request, const char* name, const char* value) { - struct json_object* params = jsonrpc_request_create_params(request); - if (!params) - return -1; - return libjson_set_string(params, name, value); + struct json_object* params = jsonrpc_request_create_params(request); + if (!params) + return -1; + return libjson_set_string(params, name, value); } int jsonrpc_request_get_param_int(const struct jsonrpc_request* request, const char* name, int64_t* value) { - struct json_object* params = NULL; - int ret = libjson_get(request->impl, jsonrpc_key_params, ¶ms); - if (ret < 0) - return ret; - return libjson_get_int(params, name, value); + struct json_object* params = NULL; + int ret = libjson_get(request->impl, jsonrpc_key_params, ¶ms); + if (ret < 0) + return ret; + return libjson_get_int(params, name, value); } int jsonrpc_request_set_param_int(struct jsonrpc_request* request, const char* name, int64_t value) { - struct json_object* params = jsonrpc_request_create_params(request); - if (!params) - return -1; - return libjson_set_int(params, name, value); + struct json_object* params = jsonrpc_request_create_params(request); + if (!params) + return -1; + return libjson_set_int(params, name, value); } const char* jsonrpc_response_to_string(const struct jsonrpc_response* response) { - return libjson_to_string_pretty(response->impl); + return libjson_to_string_pretty(response->impl); } int jsonrpc_response_create_internal(struct jsonrpc_response** _response, const struct jsonrpc_request* request, struct json_object* result, struct json_object* error) { - int ret = 0; + int ret = 0; - struct jsonrpc_response* response = malloc(sizeof(struct jsonrpc_response)); - if (!response) { - log_errno("malloc"); - ret = -1; - goto exit; - } + struct jsonrpc_response* response = malloc(sizeof(struct jsonrpc_response)); + if (!response) { + log_errno("malloc"); + ret = -1; + goto exit; + } - ret = libjson_new_object(&response->impl); - if (ret < 0) - goto free; + ret = libjson_new_object(&response->impl); + if (ret < 0) + goto free; - ret = jsonrpc_set_version(response->impl); - if (ret < 0) - goto free_impl; + ret = jsonrpc_set_version(response->impl); + if (ret < 0) + goto free_impl; - struct json_object* id = NULL; - ret = libjson_clone(request->impl, jsonrpc_key_id, &id); - if (ret < 0) - goto free_impl; + struct json_object* id = NULL; + ret = libjson_clone(request->impl, jsonrpc_key_id, &id); + if (ret < 0) + goto free_impl; - ret = libjson_set(response->impl, jsonrpc_key_id, id); - if (ret < 0) { - libjson_free(id); - goto free_impl; - } + ret = libjson_set(response->impl, jsonrpc_key_id, id); + if (ret < 0) { + libjson_free(id); + goto free_impl; + } - if (error) { - ret = libjson_set_const_key(response->impl, jsonrpc_key_error, error); - if (ret < 0) - goto free_impl; - } else { - ret = libjson_set_const_key(response->impl, jsonrpc_key_result, result); - if (ret < 0) - goto free_impl; - } + if (error) { + ret = libjson_set_const_key(response->impl, jsonrpc_key_error, error); + if (ret < 0) + goto free_impl; + } else { + ret = libjson_set_const_key(response->impl, jsonrpc_key_result, result); + if (ret < 0) + goto free_impl; + } - *_response = response; - goto exit; + *_response = response; + goto exit; free_impl: - libjson_free(response->impl); + libjson_free(response->impl); free: - free(response); + free(response); exit: - return ret; + return ret; } int jsonrpc_response_create(struct jsonrpc_response** response, const struct jsonrpc_request* request, struct json_object* result) { - return jsonrpc_response_create_internal(response, request, result, NULL); + return jsonrpc_response_create_internal(response, request, result, NULL); } void jsonrpc_response_destroy(struct jsonrpc_response* response) { - libjson_free(response->impl); - free(response); + libjson_free(response->impl); + free(response); } int jsonrpc_error_create(struct jsonrpc_response** response, struct jsonrpc_request* request, int code, const char* message) { - int ret = 0; - struct json_object* error = NULL; + int ret = 0; + struct json_object* error = NULL; - ret = libjson_new_object(&error); - if (ret < 0) - return ret; + ret = libjson_new_object(&error); + if (ret < 0) + return ret; - ret = libjson_set_int_const_key(error, jsonrpc_key_code, code); - if (ret < 0) - goto free; - ret = libjson_set_string_const_key(error, jsonrpc_key_message, message); - if (ret < 0) - goto free; + ret = libjson_set_int_const_key(error, jsonrpc_key_code, code); + if (ret < 0) + goto free; + ret = libjson_set_string_const_key(error, jsonrpc_key_message, message); + if (ret < 0) + goto free; - ret = jsonrpc_response_create_internal(response, request, NULL, error); - if (ret < 0) - goto free; + ret = jsonrpc_response_create_internal(response, request, NULL, error); + if (ret < 0) + goto free; - return ret; + return ret; free: - libjson_free(error); + libjson_free(error); - return ret; + return ret; } int jsonrpc_response_is_error(const struct jsonrpc_response* response) { - return libjson_has(response->impl, jsonrpc_key_error); + return libjson_has(response->impl, jsonrpc_key_error); } static int jsonrpc_response_from_json(struct jsonrpc_response** _response, struct json_object* impl) { - int ret = 0; + int ret = 0; - ret = jsonrpc_check_version(impl); - if (ret < 0) - return ret; - ret = jsonrpc_check_id(impl, 1); - if (ret < 0) - return ret; - ret = jsonrpc_check_result_or_error(impl); - if (ret < 0) - return ret; + ret = jsonrpc_check_version(impl); + if (ret < 0) + return ret; + ret = jsonrpc_check_id(impl, 1); + if (ret < 0) + return ret; + ret = jsonrpc_check_result_or_error(impl); + if (ret < 0) + return ret; - struct jsonrpc_response* response = malloc(sizeof(struct jsonrpc_response)); - if (!response) { - log_errno("malloc"); - return -1; - } - response->impl = impl; + struct jsonrpc_response* response = malloc(sizeof(struct jsonrpc_response)); + if (!response) { + log_errno("malloc"); + return -1; + } + response->impl = impl; - *_response = response; - return ret; + *_response = response; + return ret; } int jsonrpc_response_send(const struct jsonrpc_response* response, int fd) { - return libjson_send(response->impl, fd); + return libjson_send(response->impl, fd); } int jsonrpc_response_recv(struct jsonrpc_response** response, int fd) { - struct json_object* impl = libjson_recv(fd); - if (!impl) { - log_err("JSON-RPC: failed to receive response\n"); - return -1; - } + struct json_object* impl = libjson_recv(fd); + if (!impl) { + log_err("JSON-RPC: failed to receive response\n"); + return -1; + } - int ret = jsonrpc_response_from_json(response, impl); - if (ret < 0) - goto free_impl; + int ret = jsonrpc_response_from_json(response, impl); + if (ret < 0) + goto free_impl; - return ret; + return ret; free_impl: - libjson_free(impl); + libjson_free(impl); - return ret; + return ret; } @@ -15,36 +15,36 @@ int g_log_lvl = LOG_LVL_INFO; static inline void log_prefix_timestamp(FILE* dest) { - struct timeval tv; - struct tm tm; - char buf[64]; - size_t used = 0; + struct timeval tv; + struct tm tm; + char buf[64]; + size_t used = 0; - if (gettimeofday(&tv, NULL) < 0) - return; - if (!gmtime_r(&tv.tv_sec, &tm)) - return; + if (gettimeofday(&tv, NULL) < 0) + return; + if (!gmtime_r(&tv.tv_sec, &tm)) + return; - buf[0] = '\0'; - used += strftime(buf + used, sizeof(buf) - used, "%F %T", &tm); - long long msec = (long long)tv.tv_usec / 1000; - used += snprintf(buf + used, sizeof(buf) - used, ".%03lld | ", msec); - fprintf(dest, "%s", buf); + buf[0] = '\0'; + used += strftime(buf + used, sizeof(buf) - used, "%F %T", &tm); + long long msec = (long long)tv.tv_usec / 1000; + used += snprintf(buf + used, sizeof(buf) - used, ".%03lld | ", msec); + fprintf(dest, "%s", buf); } static inline void log_prefix_thread_id(FILE* dest) { - fprintf(dest, "%d | ", gettid()); + fprintf(dest, "%d | ", gettid()); } int log_entry_start(int lvl, FILE* dest) { - if (lvl < g_log_lvl) - return 0; - flockfile(dest); - log_prefix_timestamp(dest); - log_prefix_thread_id(dest); - return 1; + if (lvl < g_log_lvl) + return 0; + flockfile(dest); + log_prefix_timestamp(dest); + log_prefix_thread_id(dest); + return 1; } void log_entry_end(FILE* dest) { - funlockfile(dest); + funlockfile(dest); } @@ -23,64 +23,64 @@ extern int g_log_lvl; int log_entry_start(int lvl, FILE* dest); void log_entry_end(FILE* dest); -#define log_debug(...) \ - do { \ - if (!log_entry_start(LOG_LVL_DEBUG, stdout)) \ - break; \ - fprintf(stdout, __VA_ARGS__); \ - log_entry_end(stdout); \ - } while (0) +#define log_debug(...) \ + do { \ + if (!log_entry_start(LOG_LVL_DEBUG, stdout)) \ + break; \ + fprintf(stdout, __VA_ARGS__); \ + log_entry_end(stdout); \ + } while (0) -#define log(...) \ - do { \ - if (!log_entry_start(LOG_LVL_INFO, stdout)) \ - break; \ - fprintf(stdout, __VA_ARGS__); \ - log_entry_end(stdout); \ - } while (0) +#define log(...) \ + do { \ + if (!log_entry_start(LOG_LVL_INFO, stdout)) \ + break; \ + fprintf(stdout, __VA_ARGS__); \ + log_entry_end(stdout); \ + } while (0) -#define _log_err_file_loc() \ - do { \ - fprintf(stderr, "%s(%d): ", basename(__FILE__), __LINE__); \ - } while (0) +#define _log_err_file_loc() \ + do { \ + fprintf(stderr, "%s(%d): ", basename(__FILE__), __LINE__); \ + } while (0) -#define log_err(...) \ - do { \ - if (!log_entry_start(LOG_LVL_ERR, stderr)) \ - break; \ - _log_err_file_loc(); \ - fprintf(stderr, __VA_ARGS__); \ - log_entry_end(stderr); \ - } while (0) +#define log_err(...) \ + do { \ + if (!log_entry_start(LOG_LVL_ERR, stderr)) \ + break; \ + _log_err_file_loc(); \ + fprintf(stderr, __VA_ARGS__); \ + log_entry_end(stderr); \ + } while (0) -#define log_errno(fn) \ - do { \ - if (!log_entry_start(LOG_LVL_ERR, stderr)) \ - break; \ - _log_err_file_loc(); \ - perror(fn); \ - log_entry_end(stderr); \ - } while (0) +#define log_errno(fn) \ + do { \ + if (!log_entry_start(LOG_LVL_ERR, stderr)) \ + break; \ + _log_err_file_loc(); \ + perror(fn); \ + log_entry_end(stderr); \ + } while (0) -#define log_errno_if(expr, fn) \ - do { \ - int CONCAT(ret, __LINE__) = expr; \ - if (CONCAT(ret, __LINE__) < 0) \ - log_errno(fn); \ - } while (0) +#define log_errno_if(expr, fn) \ + do { \ + int CONCAT(ret, __LINE__) = expr; \ + if (CONCAT(ret, __LINE__) < 0) \ + log_errno(fn); \ + } while (0) #define pthread_errno(var, fn) \ - do { \ - errno = var; \ - log_errno(fn); \ - var = -var; \ - } while (0) + do { \ + errno = var; \ + log_errno(fn); \ + var = -var; \ + } while (0) -#define pthread_errno_if(expr, fn) \ - do { \ - int CONCAT(ret, __LINE__) = expr; \ - if (CONCAT(ret, __LINE__)) \ - pthread_errno(CONCAT(ret, __LINE__), fn); \ - } while (0) +#define pthread_errno_if(expr, fn) \ + do { \ + int CONCAT(ret, __LINE__) = expr; \ + if (CONCAT(ret, __LINE__)) \ + pthread_errno(CONCAT(ret, __LINE__), fn); \ + } while (0) #endif @@ -22,234 +22,233 @@ #define gai_log_errno(ec) log_err("getaddrinfo: %s\n", gai_strerror(ec)) int net_bind(const char* port) { - static const int flags = SOCK_CLOEXEC; - struct addrinfo *result = NULL, *it = NULL; - struct addrinfo hints; - int socket_fd = -1, ret = 0; + static const int flags = SOCK_CLOEXEC; + struct addrinfo *result = NULL, *it = NULL; + struct addrinfo hints; + int socket_fd = -1, ret = 0; - memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_INET6; - hints.ai_socktype = SOCK_STREAM; - hints.ai_flags = AI_PASSIVE; + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_INET6; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; - ret = getaddrinfo(NULL, port, &hints, &result); - if (ret) { - gai_log_errno(ret); - return -1; - } + ret = getaddrinfo(NULL, port, &hints, &result); + if (ret) { + gai_log_errno(ret); + return -1; + } - for (it = result; it; it = it->ai_next) { - socket_fd = socket(it->ai_family, it->ai_socktype | flags, it->ai_protocol); - if (socket_fd < 0) { - log_errno("socket"); - continue; - } + for (it = result; it; it = it->ai_next) { + socket_fd = socket(it->ai_family, it->ai_socktype | flags, it->ai_protocol); + if (socket_fd < 0) { + log_errno("socket"); + continue; + } - static const int yes = 1; - static const int no = 0; + static const int yes = 1; + static const int no = 0; - if (it->ai_family == AF_INET6) { - if (setsockopt(socket_fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, sizeof(no)) < 0) { - log_errno("setsockopt"); - goto close_socket; - } - } + if (it->ai_family == AF_INET6) { + if (setsockopt(socket_fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, sizeof(no)) < 0) { + log_errno("setsockopt"); + goto close_socket; + } + } - if (setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0) { - log_errno("setsockopt"); - goto close_socket; - } + if (setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0) { + log_errno("setsockopt"); + goto close_socket; + } - if (bind(socket_fd, it->ai_addr, it->ai_addrlen) < 0) { - log_errno("bind"); - goto close_socket; - } + if (bind(socket_fd, it->ai_addr, it->ai_addrlen) < 0) { + log_errno("bind"); + goto close_socket; + } - break; + break; - close_socket: - net_close(socket_fd); - } + close_socket: + net_close(socket_fd); + } - freeaddrinfo(result); + freeaddrinfo(result); - if (!it) { - log_err("Couldn't bind to port %s\n", port); - return -1; - } + if (!it) { + log_err("Couldn't bind to port %s\n", port); + return -1; + } - ret = listen(socket_fd, 4096); - if (ret < 0) { - log_errno("listen"); - goto fail; - } + ret = listen(socket_fd, 4096); + if (ret < 0) { + log_errno("listen"); + goto fail; + } - return socket_fd; + return socket_fd; fail: - net_close(socket_fd); + net_close(socket_fd); - return ret; + return ret; } int net_accept(int fd) { - static const int flags = SOCK_CLOEXEC; - int ret = 0; + static const int flags = SOCK_CLOEXEC; + int ret = 0; - ret = accept4(fd, NULL, NULL, flags); - if (ret < 0) { - log_errno("accept4"); - return ret; - } + ret = accept4(fd, NULL, NULL, flags); + if (ret < 0) { + log_errno("accept4"); + return ret; + } - return ret; + return ret; } int net_connect(const char* host, const char* port) { - static const int flags = SOCK_CLOEXEC; - struct addrinfo *result = NULL, *it = NULL; - struct addrinfo hints; - int socket_fd = -1, ret = 0; + static const int flags = SOCK_CLOEXEC; + struct addrinfo *result = NULL, *it = NULL; + struct addrinfo hints; + int socket_fd = -1, ret = 0; - memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_UNSPEC; - hints.ai_socktype = SOCK_STREAM; + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; - ret = getaddrinfo(host, port, &hints, &result); - if (ret) { - gai_log_errno(ret); - return -1; - } + ret = getaddrinfo(host, port, &hints, &result); + if (ret) { + gai_log_errno(ret); + return -1; + } - for (it = result; it; it = it->ai_next) { - socket_fd = socket(it->ai_family, it->ai_socktype | flags, it->ai_protocol); - if (socket_fd < 0) { - log_errno("socket"); - continue; - } + for (it = result; it; it = it->ai_next) { + socket_fd = socket(it->ai_family, it->ai_socktype | flags, it->ai_protocol); + if (socket_fd < 0) { + log_errno("socket"); + continue; + } - if (connect(socket_fd, it->ai_addr, it->ai_addrlen) < 0) { - log_errno("connect"); - goto close_socket; - } + if (connect(socket_fd, it->ai_addr, it->ai_addrlen) < 0) { + log_errno("connect"); + goto close_socket; + } - break; + break; - close_socket: - net_close(socket_fd); - } + close_socket: + net_close(socket_fd); + } - freeaddrinfo(result); + freeaddrinfo(result); - if (!it) { - log_err("Couldn't connect to host %s, port %s\n", host, port); - return -1; - } + if (!it) { + log_err("Couldn't connect to host %s, port %s\n", host, port); + return -1; + } - return socket_fd; + return socket_fd; } void net_close(int fd) { - file_close(fd); + file_close(fd); } static ssize_t net_send_part(int fd, const void* buf, size_t size) { - static const int flags = MSG_NOSIGNAL; + static const int flags = MSG_NOSIGNAL; - ssize_t ret = send(fd, buf, size, flags); - if (ret < 0) { - log_errno("send"); - return -1; - } + ssize_t ret = send(fd, buf, size, flags); + if (ret < 0) { + log_errno("send"); + return -1; + } - return ret; + return ret; } int net_send(int fd, const void* buf, size_t size) { - size_t sent_total = 0; + size_t sent_total = 0; - while (sent_total < size) { - ssize_t sent_now = - net_send_part(fd, (const char*)buf + sent_total, size - sent_total); - if (sent_now < 0) - return -1; - sent_total += sent_now; - } + while (sent_total < size) { + ssize_t sent_now = net_send_part(fd, (const char*)buf + sent_total, size - sent_total); + if (sent_now < 0) + return -1; + sent_total += sent_now; + } - return 0; + return 0; } int net_recv(int fd, void* buf, size_t size) { - ssize_t read_total = 0; + ssize_t read_total = 0; - while ((size_t)read_total < size) { - ssize_t read_now = read(fd, (unsigned char*)buf + read_total, size - read_total); - if (!read_now) - break; + while ((size_t)read_total < size) { + ssize_t read_now = read(fd, (unsigned char*)buf + read_total, size - read_total); + if (!read_now) + break; - if (read_now < 0) { - log_errno("read"); - return -1; - } + if (read_now < 0) { + log_errno("read"); + return -1; + } - read_total += read_now; - } + read_total += read_now; + } - if ((size_t)read_total < size) { - log_err("Received only %zd bytes out of %zu\n", read_total, size); - return -1; - } + if ((size_t)read_total < size) { + log_err("Received only %zd bytes out of %zu\n", read_total, size); + return -1; + } - return 0; + return 0; } int net_send_buf(int fd, const struct buf* buf) { - int ret = 0; + int ret = 0; - uint32_t size = htonl(buf_get_size(buf)); - ret = net_send(fd, &size, sizeof(size)); - if (ret < 0) - return ret; + uint32_t size = htonl(buf_get_size(buf)); + ret = net_send(fd, &size, sizeof(size)); + if (ret < 0) + return ret; - ret = net_send(fd, buf_get_data(buf), buf_get_size(buf)); - if (ret < 0) - return ret; + ret = net_send(fd, buf_get_data(buf), buf_get_size(buf)); + if (ret < 0) + return ret; - return ret; + return ret; } int net_recv_buf(int fd, struct buf** buf) { - uint32_t size = 0; - int ret = 0; + uint32_t size = 0; + int ret = 0; - ret = net_recv(fd, &size, sizeof(size)); - if (ret < 0) { - log_err("Couldn't read buffer size\n"); - goto fail; - } - size = ntohl(size); + ret = net_recv(fd, &size, sizeof(size)); + if (ret < 0) { + log_err("Couldn't read buffer size\n"); + goto fail; + } + size = ntohl(size); - void* data = malloc(size); - if (!data) { - log_errno("malloc"); - goto fail; - } + void* data = malloc(size); + if (!data) { + log_errno("malloc"); + goto fail; + } - ret = net_recv(fd, data, size); - if (ret < 0) { - log_err("Couldn't read buffer\n"); - goto free_data; - } + ret = net_recv(fd, data, size); + if (ret < 0) { + log_err("Couldn't read buffer\n"); + goto free_data; + } - ret = buf_create(buf, data, size); - if (ret < 0) - goto free_data; + ret = buf_create(buf, data, size); + if (ret < 0) + goto free_data; - return ret; + return ret; free_data: - free(data); + free(data); fail: - return -1; + return -1; } diff --git a/src/process.c b/src/process.c index 8994de0..ea93904 100644 --- a/src/process.c +++ b/src/process.c @@ -16,146 +16,146 @@ #include <unistd.h> static int exec_child(const char* args[], const char* envp[]) { - static const char* default_envp[] = {NULL}; + static const char* default_envp[] = {NULL}; - if (!envp) - envp = default_envp; + if (!envp) + envp = default_envp; - int ret = execvpe(args[0], (char* const*)args, (char* const*)envp); - if (ret < 0) { - log_errno("execvpe"); - return ret; - } + int ret = execvpe(args[0], (char* const*)args, (char* const*)envp); + if (ret < 0) { + log_errno("execvpe"); + return ret; + } - return ret; + return ret; } static int wait_for_child(pid_t pid, int* ec) { - int status; + int status; - pid_t ret = waitpid(pid, &status, __WNOTHREAD); - if (ret < 0) { - log_errno("waitpid"); - return ret; - } + pid_t ret = waitpid(pid, &status, __WNOTHREAD); + if (ret < 0) { + log_errno("waitpid"); + return ret; + } - /* The child process reports the lowest 8 bits of its exit code, which - * are treated as an unsigned integer on Linux. - * - * If it was killed by a signal, indicate that by negating the signal - * number. */ + /* The child process reports the lowest 8 bits of its exit code, which + * are treated as an unsigned integer on Linux. + * + * If it was killed by a signal, indicate that by negating the signal + * number. */ - if (WIFEXITED(status)) { - *ec = WEXITSTATUS(status); - } else if (WIFSIGNALED(status)) { - *ec = -WTERMSIG(status); - } else { - log_err("This shouldn't happen: %d\n", status); - *ec = 1; - } + if (WIFEXITED(status)) { + *ec = WEXITSTATUS(status); + } else if (WIFSIGNALED(status)) { + *ec = -WTERMSIG(status); + } else { + log_err("This shouldn't happen: %d\n", status); + *ec = 1; + } - return 0; + return 0; } int process_execute(const char* args[], const char* envp[], int* ec) { - pid_t child_pid = fork(); - if (child_pid < 0) { - log_errno("fork"); - return child_pid; - } + pid_t child_pid = fork(); + if (child_pid < 0) { + log_errno("fork"); + return child_pid; + } - if (!child_pid) - exit(exec_child(args, envp)); + if (!child_pid) + exit(exec_child(args, envp)); - return wait_for_child(child_pid, ec); + return wait_for_child(child_pid, ec); } static int redirect_and_exec_child(int pipe_fds[2], const char* args[], const char* envp[]) { - int ret = 0; + int ret = 0; - file_close(pipe_fds[0]); + file_close(pipe_fds[0]); - ret = dup2(pipe_fds[1], STDOUT_FILENO); - if (ret < 0) { - log_errno("dup2"); - return ret; - } + ret = dup2(pipe_fds[1], STDOUT_FILENO); + if (ret < 0) { + log_errno("dup2"); + return ret; + } - ret = dup2(pipe_fds[1], STDERR_FILENO); - if (ret < 0) { - log_errno("dup2"); - return ret; - } + ret = dup2(pipe_fds[1], STDERR_FILENO); + if (ret < 0) { + log_errno("dup2"); + return ret; + } - return exec_child(args, envp); + return exec_child(args, envp); } int process_execute_and_capture(const char* args[], const char* envp[], struct process_output* result) { - static const int flags = O_CLOEXEC; - int pipe_fds[2]; - int ret = 0; + static const int flags = O_CLOEXEC; + int pipe_fds[2]; + int ret = 0; - ret = pipe2(pipe_fds, flags); - if (ret < 0) { - log_errno("pipe2"); - return -1; - } + ret = pipe2(pipe_fds, flags); + if (ret < 0) { + log_errno("pipe2"); + return -1; + } - pid_t child_pid = fork(); - if (child_pid < 0) { - log_errno("fork"); - goto close_pipe; - } + pid_t child_pid = fork(); + if (child_pid < 0) { + log_errno("fork"); + goto close_pipe; + } - if (!child_pid) - exit(redirect_and_exec_child(pipe_fds, args, envp)); + if (!child_pid) + exit(redirect_and_exec_child(pipe_fds, args, envp)); - file_close(pipe_fds[1]); + file_close(pipe_fds[1]); - ret = file_read(pipe_fds[0], &result->data, &result->data_size); - if (ret < 0) - goto close_pipe; + ret = file_read(pipe_fds[0], &result->data, &result->data_size); + if (ret < 0) + goto close_pipe; - ret = wait_for_child(child_pid, &result->ec); - if (ret < 0) - goto free_data; + ret = wait_for_child(child_pid, &result->ec); + if (ret < 0) + goto free_data; - goto close_pipe; + goto close_pipe; free_data: - free(result->data); + free(result->data); close_pipe: - file_close(pipe_fds[0]); - /* No errno checking here, we might've already closed the write end. */ - close(pipe_fds[1]); + file_close(pipe_fds[0]); + /* No errno checking here, we might've already closed the write end. */ + close(pipe_fds[1]); - return ret; + return ret; } int process_output_create(struct process_output** _output) { - struct process_output* output = calloc(1, sizeof(struct process_output)); - if (!output) { - log_errno("calloc"); - return -1; - } + struct process_output* output = calloc(1, sizeof(struct process_output)); + if (!output) { + log_errno("calloc"); + return -1; + } - output->ec = 0; - output->data = NULL; - output->data_size = 0; + output->ec = 0; + output->data = NULL; + output->data_size = 0; - *_output = output; - return 0; + *_output = output; + return 0; } void process_output_destroy(struct process_output* output) { - free(output->data); - free(output); + free(output->data); + free(output); } void process_output_dump(const struct process_output* output) { - log("Process exit code: %d\n", output->ec); - log("Process output: %zu bytes\n", output->data_size); + log("Process exit code: %d\n", output->ec); + log("Process output: %zu bytes\n", output->data_size); } diff --git a/src/process.h b/src/process.h index 00b68f9..b3ee603 100644 --- a/src/process.h +++ b/src/process.h @@ -11,9 +11,9 @@ #include <stddef.h> struct process_output { - int ec; - unsigned char* data; - size_t data_size; + int ec; + unsigned char* data; + size_t data_size; }; /* The exit code is only valid if the functions returns a non-negative number. */ diff --git a/src/protocol.c b/src/protocol.c index e650c49..dc72434 100644 --- a/src/protocol.c +++ b/src/protocol.c @@ -24,90 +24,90 @@ static const char* const run_key_url = "url"; static const char* const run_key_rev = "rev"; int request_create_queue_run(struct jsonrpc_request** request, const struct run* run) { - int ret = 0; + int ret = 0; - ret = jsonrpc_request_create(request, jsonrpc_generate_request_id(), CMD_QUEUE_RUN, NULL); - if (ret < 0) - return ret; - ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_repo_url(run)); - if (ret < 0) - goto free_request; - ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_repo_rev(run)); - if (ret < 0) - goto free_request; + ret = jsonrpc_request_create(request, jsonrpc_generate_request_id(), CMD_QUEUE_RUN, NULL); + if (ret < 0) + return ret; + ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_repo_url(run)); + if (ret < 0) + goto free_request; + ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_repo_rev(run)); + if (ret < 0) + goto free_request; - return ret; + return ret; free_request: - jsonrpc_request_destroy(*request); + jsonrpc_request_destroy(*request); - return ret; + return ret; } int request_parse_queue_run(const struct jsonrpc_request* request, struct run** run) { - int ret = 0; + int ret = 0; - const char* url = NULL; - ret = jsonrpc_request_get_param_string(request, run_key_url, &url); - if (ret < 0) - return ret; - const char* rev = NULL; - ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev); - if (ret < 0) - return ret; + const char* url = NULL; + ret = jsonrpc_request_get_param_string(request, run_key_url, &url); + if (ret < 0) + return ret; + const char* rev = NULL; + ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev); + if (ret < 0) + return ret; - return run_queued(run, url, rev); + return run_queued(run, url, rev); } int request_create_new_worker(struct jsonrpc_request** request) { - return jsonrpc_notification_create(request, CMD_NEW_WORKER, NULL); + return jsonrpc_notification_create(request, CMD_NEW_WORKER, NULL); } int request_parse_new_worker(UNUSED const struct jsonrpc_request* request) { - return 0; + return 0; } int request_create_start_run(struct jsonrpc_request** request, const struct run* run) { - int ret = 0; + int ret = 0; - ret = jsonrpc_notification_create(request, CMD_START_RUN, NULL); - if (ret < 0) - return ret; - ret = jsonrpc_request_set_param_int(*request, run_key_id, run_get_id(run)); - if (ret < 0) - goto free_request; - ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_repo_url(run)); - if (ret < 0) - goto free_request; - ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_repo_rev(run)); - if (ret < 0) - goto free_request; + ret = jsonrpc_notification_create(request, CMD_START_RUN, NULL); + if (ret < 0) + return ret; + ret = jsonrpc_request_set_param_int(*request, run_key_id, run_get_id(run)); + if (ret < 0) + goto free_request; + ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_repo_url(run)); + if (ret < 0) + goto free_request; + ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_repo_rev(run)); + if (ret < 0) + goto free_request; - return ret; + return ret; free_request: - jsonrpc_request_destroy(*request); + jsonrpc_request_destroy(*request); - return ret; + return ret; } int request_parse_start_run(const struct jsonrpc_request* request, struct run** run) { - int ret = 0; + int ret = 0; - int64_t id = 0; - ret = jsonrpc_request_get_param_int(request, run_key_id, &id); - if (ret < 0) - return ret; - const char* url = NULL; - ret = jsonrpc_request_get_param_string(request, run_key_url, &url); - if (ret < 0) - return ret; - const char* rev = NULL; - ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev); - if (ret < 0) - return ret; + int64_t id = 0; + ret = jsonrpc_request_get_param_int(request, run_key_id, &id); + if (ret < 0) + return ret; + const char* url = NULL; + ret = jsonrpc_request_get_param_string(request, run_key_url, &url); + if (ret < 0) + return ret; + const char* rev = NULL; + ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev); + if (ret < 0) + return ret; - return run_created(run, (int)id, url, rev); + return run_created(run, (int)id, url, rev); } static const char* const finished_key_run_id = "run_id"; @@ -117,102 +117,102 @@ static const char* const finished_key_data = "output"; int request_create_finished_run(struct jsonrpc_request** request, int run_id, const struct process_output* output) { - int ret = 0; + int ret = 0; - ret = jsonrpc_notification_create(request, CMD_FINISHED_RUN, NULL); - if (ret < 0) - return ret; - ret = jsonrpc_request_set_param_int(*request, finished_key_run_id, run_id); - if (ret < 0) - goto free_request; - ret = jsonrpc_request_set_param_int(*request, finished_key_ec, output->ec); - if (ret < 0) - goto free_request; + ret = jsonrpc_notification_create(request, CMD_FINISHED_RUN, NULL); + if (ret < 0) + return ret; + ret = jsonrpc_request_set_param_int(*request, finished_key_run_id, run_id); + if (ret < 0) + goto free_request; + ret = jsonrpc_request_set_param_int(*request, finished_key_ec, output->ec); + if (ret < 0) + goto free_request; - char* b64data = NULL; - ret = base64_encode(output->data, output->data_size, &b64data); - if (ret < 0) - goto free_request; + char* b64data = NULL; + ret = base64_encode(output->data, output->data_size, &b64data); + if (ret < 0) + goto free_request; - ret = jsonrpc_request_set_param_string(*request, finished_key_data, b64data); - free(b64data); - if (ret < 0) - goto free_request; + ret = jsonrpc_request_set_param_string(*request, finished_key_data, b64data); + free(b64data); + if (ret < 0) + goto free_request; - return ret; + return ret; free_request: - jsonrpc_request_destroy(*request); + jsonrpc_request_destroy(*request); - return ret; + return ret; } int request_parse_finished_run(const struct jsonrpc_request* request, int* _run_id, struct process_output** _output) { - int ret = 0; + int ret = 0; - struct process_output* output = NULL; - ret = process_output_create(&output); - if (ret < 0) - return ret; + struct process_output* output = NULL; + ret = process_output_create(&output); + if (ret < 0) + return ret; - int64_t run_id = 0; - ret = jsonrpc_request_get_param_int(request, finished_key_run_id, &run_id); - if (ret < 0) - goto free_output; + int64_t run_id = 0; + ret = jsonrpc_request_get_param_int(request, finished_key_run_id, &run_id); + if (ret < 0) + goto free_output; - int64_t ec = -1; - ret = jsonrpc_request_get_param_int(request, finished_key_ec, &ec); - if (ret < 0) - goto free_output; - output->ec = (int)ec; + int64_t ec = -1; + ret = jsonrpc_request_get_param_int(request, finished_key_ec, &ec); + if (ret < 0) + goto free_output; + output->ec = (int)ec; - const char* b64data = NULL; - ret = jsonrpc_request_get_param_string(request, finished_key_data, &b64data); - if (ret < 0) - goto free_output; + const char* b64data = NULL; + ret = jsonrpc_request_get_param_string(request, finished_key_data, &b64data); + if (ret < 0) + goto free_output; - ret = base64_decode(b64data, &output->data, &output->data_size); - if (ret < 0) - goto free_output; + ret = base64_decode(b64data, &output->data, &output->data_size); + if (ret < 0) + goto free_output; - *_run_id = (int)run_id; - *_output = output; - return ret; + *_run_id = (int)run_id; + *_output = output; + return ret; free_output: - process_output_destroy(output); + process_output_destroy(output); - return ret; + return ret; } int request_create_get_runs(struct jsonrpc_request** request) { - return jsonrpc_request_create(request, jsonrpc_generate_request_id(), CMD_GET_RUNS, NULL); + return jsonrpc_request_create(request, jsonrpc_generate_request_id(), CMD_GET_RUNS, NULL); } int request_parse_get_runs(UNUSED const struct jsonrpc_request* request) { - return 0; + return 0; } int response_create_get_runs(struct jsonrpc_response** response, const struct jsonrpc_request* request, const struct run_queue* runs) { - struct json_object* runs_json = NULL; - int ret = 0; + struct json_object* runs_json = NULL; + int ret = 0; - ret = run_queue_to_json(runs, &runs_json); - if (ret < 0) - return ret; + ret = run_queue_to_json(runs, &runs_json); + if (ret < 0) + return ret; - ret = jsonrpc_response_create(response, request, runs_json); - if (ret < 0) - goto free_json; + ret = jsonrpc_response_create(response, request, runs_json); + if (ret < 0) + goto free_json; - return ret; + return ret; free_json: - libjson_free(runs_json); + libjson_free(runs_json); - return ret; + return ret; } diff --git a/src/run_queue.c b/src/run_queue.c index 652096a..b6093e4 100644 --- a/src/run_queue.c +++ b/src/run_queue.c @@ -17,13 +17,13 @@ #include <sys/queue.h> struct run { - int id; - char* repo_url; - char* repo_rev; - int status; - int exit_code; + int id; + char* repo_url; + char* repo_rev; + int status; + int exit_code; - SIMPLEQ_ENTRY(run) entries; + SIMPLEQ_ENTRY(run) entries; }; int run_new(struct run** _entry, @@ -32,161 +32,161 @@ int run_new(struct run** _entry, const char* _repo_rev, enum run_status status, int exit_code) { - struct run* entry = malloc(sizeof(struct run)); - if (!entry) { - log_errno("malloc"); - goto fail; - } + struct run* entry = malloc(sizeof(struct run)); + if (!entry) { + log_errno("malloc"); + goto fail; + } - char* repo_url = strdup(_repo_url); - if (!repo_url) { - log_errno("strdup"); - goto free_entry; - } + char* repo_url = strdup(_repo_url); + if (!repo_url) { + log_errno("strdup"); + goto free_entry; + } - char* repo_rev = strdup(_repo_rev); - if (!repo_rev) { - log_errno("strdup"); - goto free_repo_url; - } + char* repo_rev = strdup(_repo_rev); + if (!repo_rev) { + log_errno("strdup"); + goto free_repo_url; + } - entry->id = id; - entry->repo_url = repo_url; - entry->repo_rev = repo_rev; - entry->status = status; - entry->exit_code = exit_code; + entry->id = id; + entry->repo_url = repo_url; + entry->repo_rev = repo_rev; + entry->status = status; + entry->exit_code = exit_code; - *_entry = entry; - return 0; + *_entry = entry; + return 0; free_repo_url: - free(repo_url); + free(repo_url); free_entry: - free(entry); + free(entry); fail: - return -1; + return -1; } void run_destroy(struct run* entry) { - free(entry->repo_rev); - free(entry->repo_url); - free(entry); + free(entry->repo_rev); + free(entry->repo_url); + free(entry); } int run_queued(struct run** entry, const char* repo_url, const char* repo_rev) { - return run_new(entry, -1, repo_url, repo_rev, RUN_STATUS_CREATED, -1); + return run_new(entry, -1, repo_url, repo_rev, RUN_STATUS_CREATED, -1); } int run_created(struct run** entry, int id, const char* repo_url, const char* repo_rev) { - return run_new(entry, id, repo_url, repo_rev, RUN_STATUS_CREATED, -1); + return run_new(entry, id, repo_url, repo_rev, RUN_STATUS_CREATED, -1); } int run_to_json(const struct run* entry, struct json_object** _json) { - struct json_object* json = NULL; - int ret = 0; + struct json_object* json = NULL; + int ret = 0; - ret = libjson_new_object(&json); - if (ret < 0) - return -1; - ret = libjson_set_int_const_key(json, "id", entry->id); - if (ret < 0) - goto free; - ret = libjson_set_int_const_key(json, "exit_code", entry->exit_code); - if (ret < 0) - goto free; - ret = libjson_set_string_const_key(json, "repo_url", entry->repo_url); - if (ret < 0) - goto free; - ret = libjson_set_string_const_key(json, "repo_rev", entry->repo_rev); - if (ret < 0) - goto free; + ret = libjson_new_object(&json); + if (ret < 0) + return -1; + ret = libjson_set_int_const_key(json, "id", entry->id); + if (ret < 0) + goto free; + ret = libjson_set_int_const_key(json, "exit_code", entry->exit_code); + if (ret < 0) + goto free; + ret = libjson_set_string_const_key(json, "repo_url", entry->repo_url); + if (ret < 0) + goto free; + ret = libjson_set_string_const_key(json, "repo_rev", entry->repo_rev); + if (ret < 0) + goto free; - *_json = json; - return ret; + *_json = json; + return ret; free: - libjson_free(json); + libjson_free(json); - return ret; + return ret; } int run_get_id(const struct run* entry) { - return entry->id; + return entry->id; } const char* run_get_repo_url(const struct run* entry) { - return entry->repo_url; + return entry->repo_url; } const char* run_get_repo_rev(const struct run* entry) { - return entry->repo_rev; + return entry->repo_rev; } void run_set_id(struct run* entry, int id) { - entry->id = id; + entry->id = id; } void run_queue_create(struct run_queue* queue) { - SIMPLEQ_INIT(queue); + SIMPLEQ_INIT(queue); } void run_queue_destroy(struct run_queue* queue) { - struct run* entry1 = SIMPLEQ_FIRST(queue); - while (entry1) { - struct run* entry2 = SIMPLEQ_NEXT(entry1, entries); - run_destroy(entry1); - entry1 = entry2; - } - SIMPLEQ_INIT(queue); + struct run* entry1 = SIMPLEQ_FIRST(queue); + while (entry1) { + struct run* entry2 = SIMPLEQ_NEXT(entry1, entries); + run_destroy(entry1); + entry1 = entry2; + } + SIMPLEQ_INIT(queue); } int run_queue_to_json(const struct run_queue* queue, struct json_object** _json) { - struct json_object* json = NULL; - int ret = 0; + struct json_object* json = NULL; + int ret = 0; - ret = libjson_new_array(&json); - if (ret < 0) - return ret; + ret = libjson_new_array(&json); + if (ret < 0) + return ret; - struct run* entry = NULL; - SIMPLEQ_FOREACH(entry, queue, entries) { - struct json_object* entry_json = NULL; - ret = run_to_json(entry, &entry_json); - if (ret < 0) - goto free; + struct run* entry = NULL; + SIMPLEQ_FOREACH(entry, queue, entries) { + struct json_object* entry_json = NULL; + ret = run_to_json(entry, &entry_json); + if (ret < 0) + goto free; - ret = libjson_append(json, entry_json); - if (ret < 0) { - libjson_free(entry_json); - goto free; - } - } + ret = libjson_append(json, entry_json); + if (ret < 0) { + libjson_free(entry_json); + goto free; + } + } - *_json = json; - return ret; + *_json = json; + return ret; free: - libjson_free(json); + libjson_free(json); - return ret; + return ret; } int run_queue_is_empty(const struct run_queue* queue) { - return SIMPLEQ_EMPTY(queue); + return SIMPLEQ_EMPTY(queue); } void run_queue_add_first(struct run_queue* queue, struct run* entry) { - SIMPLEQ_INSERT_HEAD(queue, entry, entries); + SIMPLEQ_INSERT_HEAD(queue, entry, entries); } void run_queue_add_last(struct run_queue* queue, struct run* entry) { - SIMPLEQ_INSERT_TAIL(queue, entry, entries); + SIMPLEQ_INSERT_TAIL(queue, entry, entries); } struct run* run_queue_remove_first(struct run_queue* queue) { - struct run* entry = SIMPLEQ_FIRST(queue); - SIMPLEQ_REMOVE_HEAD(queue, entries); - return entry; + struct run* entry = SIMPLEQ_FIRST(queue); + SIMPLEQ_REMOVE_HEAD(queue, entries); + return entry; } diff --git a/src/run_queue.h b/src/run_queue.h index 406227e..92948a9 100644 --- a/src/run_queue.h +++ b/src/run_queue.h @@ -13,8 +13,8 @@ #include <sys/queue.h> enum run_status { - RUN_STATUS_CREATED = 1, - RUN_STATUS_FINISHED = 2, + RUN_STATUS_CREATED = 1, + RUN_STATUS_FINISHED = 2, }; struct run; diff --git a/src/server.c b/src/server.c index 264a165..dabc0f4 100644 --- a/src/server.c +++ b/src/server.c @@ -29,316 +29,316 @@ #include <stdlib.h> struct server { - pthread_mutex_t server_mtx; - pthread_cond_t server_cv; + pthread_mutex_t server_mtx; + pthread_cond_t server_cv; - int stopping; + int stopping; - struct cmd_dispatcher* cmd_dispatcher; + struct cmd_dispatcher* cmd_dispatcher; - struct event_loop* event_loop; - int signalfd; + struct event_loop* event_loop; + int signalfd; - struct worker_queue worker_queue; - struct run_queue run_queue; + struct worker_queue worker_queue; + struct run_queue run_queue; - struct storage storage; + struct storage storage; - pthread_t main_thread; + pthread_t main_thread; - struct tcp_server* tcp_server; + struct tcp_server* tcp_server; }; static int server_lock(struct server* server) { - int ret = pthread_mutex_lock(&server->server_mtx); - if (ret) { - pthread_errno(ret, "pthread_mutex_lock"); - return ret; - } - return ret; + int ret = pthread_mutex_lock(&server->server_mtx); + if (ret) { + pthread_errno(ret, "pthread_mutex_lock"); + return ret; + } + return ret; } static void server_unlock(struct server* server) { - pthread_errno_if(pthread_mutex_unlock(&server->server_mtx), "pthread_mutex_unlock"); + pthread_errno_if(pthread_mutex_unlock(&server->server_mtx), "pthread_mutex_unlock"); } static int server_wait(struct server* server) { - int ret = pthread_cond_wait(&server->server_cv, &server->server_mtx); - if (ret) { - pthread_errno(ret, "pthread_cond_wait"); - return ret; - } - return ret; + int ret = pthread_cond_wait(&server->server_cv, &server->server_mtx); + if (ret) { + pthread_errno(ret, "pthread_cond_wait"); + return ret; + } + return ret; } static void server_notify(struct server* server) { - pthread_errno_if(pthread_cond_signal(&server->server_cv), "pthread_cond_signal"); + pthread_errno_if(pthread_cond_signal(&server->server_cv), "pthread_cond_signal"); } static int server_set_stopping(UNUSED struct event_loop* loop, UNUSED int fd, UNUSED short revents, void* _server) { - struct server* server = (struct server*)_server; - int ret = 0; + struct server* server = (struct server*)_server; + int ret = 0; - ret = server_lock(server); - if (ret < 0) - return ret; + ret = server_lock(server); + if (ret < 0) + return ret; - server->stopping = 1; + server->stopping = 1; - server_notify(server); - server_unlock(server); - return ret; + server_notify(server); + server_unlock(server); + return ret; } static int server_has_workers(const struct server* server) { - return !worker_queue_is_empty(&server->worker_queue); + return !worker_queue_is_empty(&server->worker_queue); } static int server_enqueue_worker(struct server* server, struct worker* worker) { - int ret = 0; + int ret = 0; - ret = server_lock(server); - if (ret < 0) - return ret; + ret = server_lock(server); + if (ret < 0) + return ret; - worker_queue_add_last(&server->worker_queue, worker); - log("Added a new worker %d to the queue\n", worker_get_fd(worker)); + worker_queue_add_last(&server->worker_queue, worker); + log("Added a new worker %d to the queue\n", worker_get_fd(worker)); - server_notify(server); - server_unlock(server); - return ret; + server_notify(server); + server_unlock(server); + return ret; } static int server_has_runs(const struct server* server) { - return !run_queue_is_empty(&server->run_queue); + return !run_queue_is_empty(&server->run_queue); } static int server_enqueue_run(struct server* server, struct run* run) { - int ret = 0; + int ret = 0; - ret = storage_run_create(&server->storage, run_get_repo_url(run), run_get_repo_rev(run)); - if (ret < 0) - return ret; - run_set_id(run, ret); + ret = storage_run_create(&server->storage, run_get_repo_url(run), run_get_repo_rev(run)); + if (ret < 0) + return ret; + run_set_id(run, ret); - ret = server_lock(server); - if (ret < 0) - return ret; + ret = server_lock(server); + if (ret < 0) + return ret; - run_queue_add_last(&server->run_queue, run); - log("Added a new run %d for repository %s to the queue\n", - run_get_id(run), - run_get_repo_url(run)); + run_queue_add_last(&server->run_queue, run); + log("Added a new run %d for repository %s to the queue\n", + run_get_id(run), + run_get_repo_url(run)); - server_notify(server); - server_unlock(server); - return ret; + server_notify(server); + server_unlock(server); + return ret; } static int server_ready_for_action(const struct server* server) { - return server->stopping || (server_has_runs(server) && server_has_workers(server)); + return server->stopping || (server_has_runs(server) && server_has_workers(server)); } static int server_wait_for_action(struct server* server) { - int ret = 0; + int ret = 0; - while (!server_ready_for_action(server)) { - ret = server_wait(server); - if (ret < 0) - return ret; - } + while (!server_ready_for_action(server)) { + ret = server_wait(server); + if (ret < 0) + return ret; + } - return ret; + return ret; } static void server_assign_run(struct server* server) { - struct run* run = run_queue_remove_first(&server->run_queue); - log("Removed run %d for repository %s from the queue\n", - run_get_id(run), - run_get_repo_url(run)); + struct run* run = run_queue_remove_first(&server->run_queue); + log("Removed run %d for repository %s from the queue\n", + run_get_id(run), + run_get_repo_url(run)); - struct worker* worker = worker_queue_remove_first(&server->worker_queue); - log("Removed worker %d from the queue\n", worker_get_fd(worker)); + struct worker* worker = worker_queue_remove_first(&server->worker_queue); + log("Removed worker %d from the queue\n", worker_get_fd(worker)); - struct jsonrpc_request* start_request = NULL; - int ret = 0; + struct jsonrpc_request* start_request = NULL; + int ret = 0; - ret = request_create_start_run(&start_request, run); - if (ret < 0) - goto exit; + ret = request_create_start_run(&start_request, run); + if (ret < 0) + goto exit; - ret = jsonrpc_request_send(start_request, worker_get_fd(worker)); - jsonrpc_request_destroy(start_request); - if (ret < 0) - goto exit; + ret = jsonrpc_request_send(start_request, worker_get_fd(worker)); + jsonrpc_request_destroy(start_request); + if (ret < 0) + goto exit; exit: - if (ret < 0) { - log("Failed to assign run for repository %s to worker %d, requeueing\n", - run_get_repo_url(run), - worker_get_fd(worker)); - run_queue_add_first(&server->run_queue, run); - } else { - log("Assigned run %d for repository %s to worker %d\n", - run_get_id(run), - run_get_repo_url(run), - worker_get_fd(worker)); - run_destroy(run); - } + if (ret < 0) { + log("Failed to assign run for repository %s to worker %d, requeueing\n", + run_get_repo_url(run), + worker_get_fd(worker)); + run_queue_add_first(&server->run_queue, run); + } else { + log("Assigned run %d for repository %s to worker %d\n", + run_get_id(run), + run_get_repo_url(run), + worker_get_fd(worker)); + run_destroy(run); + } - worker_destroy(worker); + worker_destroy(worker); } static void* server_main_thread(void* _server) { - struct server* server = (struct server*)_server; - int ret = 0; + struct server* server = (struct server*)_server; + int ret = 0; - ret = server_lock(server); - if (ret < 0) - goto exit; + ret = server_lock(server); + if (ret < 0) + goto exit; - while (1) { - ret = server_wait_for_action(server); - if (ret < 0) - goto unlock; + while (1) { + ret = server_wait_for_action(server); + if (ret < 0) + goto unlock; - if (server->stopping) - goto unlock; + if (server->stopping) + goto unlock; - server_assign_run(server); - } + server_assign_run(server); + } unlock: - server_unlock(server); + server_unlock(server); exit: - return NULL; + return NULL; } static int server_handle_cmd_new_worker(UNUSED const struct jsonrpc_request* request, UNUSED struct jsonrpc_response** response, void* _ctx) { - struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; - struct server* server = (struct server*)ctx->arg; - int ret = 0; + struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; + struct server* server = (struct server*)ctx->arg; + int ret = 0; - ret = file_dup(ctx->fd); - if (ret < 0) - return ret; + ret = file_dup(ctx->fd); + if (ret < 0) + return ret; - const int fd = ret; - struct worker* worker = NULL; + const int fd = ret; + struct worker* worker = NULL; - ret = worker_create(&worker, fd); - if (ret < 0) - goto close; + ret = worker_create(&worker, fd); + if (ret < 0) + goto close; - ret = server_enqueue_worker(server, worker); - if (ret < 0) - goto destroy_worker; + ret = server_enqueue_worker(server, worker); + if (ret < 0) + goto destroy_worker; - return ret; + return ret; destroy_worker: - worker_destroy(worker); + worker_destroy(worker); close: - net_close(fd); + net_close(fd); - return ret; + return ret; } static int server_handle_cmd_queue_run(const struct jsonrpc_request* request, struct jsonrpc_response** response, void* _ctx) { - struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; - struct server* server = (struct server*)ctx->arg; - int ret = 0; + struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; + struct server* server = (struct server*)ctx->arg; + int ret = 0; - struct run* run = NULL; + struct run* run = NULL; - ret = request_parse_queue_run(request, &run); - if (ret < 0) - return ret; + ret = request_parse_queue_run(request, &run); + if (ret < 0) + return ret; - ret = jsonrpc_response_create(response, request, NULL); - if (ret < 0) - goto destroy_run; + ret = jsonrpc_response_create(response, request, NULL); + if (ret < 0) + goto destroy_run; - ret = server_enqueue_run(server, run); - if (ret < 0) - goto free_response; + ret = server_enqueue_run(server, run); + if (ret < 0) + goto free_response; - return ret; + return ret; free_response: - jsonrpc_response_destroy(*response); - *response = NULL; + jsonrpc_response_destroy(*response); + *response = NULL; destroy_run: - run_destroy(run); + run_destroy(run); - return ret; + return ret; } static int server_handle_cmd_finished_run(const struct jsonrpc_request* request, UNUSED struct jsonrpc_response** response, void* _ctx) { - struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; - struct server* server = (struct server*)ctx->arg; - int ret = 0; + struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; + struct server* server = (struct server*)ctx->arg; + int ret = 0; - int run_id = 0; - struct process_output* output; + int run_id = 0; + struct process_output* output; - ret = request_parse_finished_run(request, &run_id, &output); - if (ret < 0) - return ret; + ret = request_parse_finished_run(request, &run_id, &output); + if (ret < 0) + return ret; - ret = storage_run_finished(&server->storage, run_id, output); - if (ret < 0) { - log_err("Failed to mark run %d as finished\n", run_id); - goto free_output; - } + ret = storage_run_finished(&server->storage, run_id, output); + if (ret < 0) { + log_err("Failed to mark run %d as finished\n", run_id); + goto free_output; + } - log("Marked run %d as finished\n", run_id); + log("Marked run %d as finished\n", run_id); free_output: - process_output_destroy(output); + process_output_destroy(output); - return ret; + return ret; } static int server_handle_cmd_get_runs(const struct jsonrpc_request* request, struct jsonrpc_response** response, void* _ctx) { - struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; - struct server* server = (struct server*)ctx->arg; - int ret = 0; + struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; + struct server* server = (struct server*)ctx->arg; + int ret = 0; - ret = request_parse_get_runs(request); - if (ret < 0) - return ret; + ret = request_parse_get_runs(request); + if (ret < 0) + return ret; - struct run_queue runs; + struct run_queue runs; - ret = storage_get_runs(&server->storage, &runs); - if (ret < 0) { - log_err("Failed to fetch runs\n"); - return ret; - } + ret = storage_get_runs(&server->storage, &runs); + if (ret < 0) { + log_err("Failed to fetch runs\n"); + return ret; + } - ret = response_create_get_runs(response, request, &runs); - if (ret < 0) - goto destroy_runs; + ret = response_create_get_runs(response, request, &runs); + if (ret < 0) + goto destroy_runs; destroy_runs: - run_queue_destroy(&runs); + run_queue_destroy(&runs); - return ret; + return ret; } static struct cmd_desc commands[] = { @@ -351,142 +351,141 @@ static struct cmd_desc commands[] = { static const size_t numof_commands = sizeof(commands) / sizeof(commands[0]); int server_create(struct server** _server, const struct settings* settings) { - struct storage_settings storage_settings; - int ret = 0; + struct storage_settings storage_settings; + int ret = 0; - struct server* server = malloc(sizeof(struct server)); - if (!server) { - log_errno("malloc"); - return -1; - } + struct server* server = malloc(sizeof(struct server)); + if (!server) { + log_errno("malloc"); + return -1; + } - ret = pthread_mutex_init(&server->server_mtx, NULL); - if (ret) { - pthread_errno(ret, "pthread_mutex_init"); - goto free; - } + ret = pthread_mutex_init(&server->server_mtx, NULL); + if (ret) { + pthread_errno(ret, "pthread_mutex_init"); + goto free; + } - ret = pthread_cond_init(&server->server_cv, NULL); - if (ret) { - pthread_errno(ret, "pthread_cond_init"); - goto destroy_mtx; - } + ret = pthread_cond_init(&server->server_cv, NULL); + if (ret) { + pthread_errno(ret, "pthread_cond_init"); + goto destroy_mtx; + } - server->stopping = 0; + server->stopping = 0; - ret = cmd_dispatcher_create(&server->cmd_dispatcher, commands, numof_commands, server); - if (ret < 0) - goto destroy_cv; + ret = cmd_dispatcher_create(&server->cmd_dispatcher, commands, numof_commands, server); + if (ret < 0) + goto destroy_cv; - ret = event_loop_create(&server->event_loop); - if (ret < 0) - goto destroy_cmd_dispatcher; + ret = event_loop_create(&server->event_loop); + if (ret < 0) + goto destroy_cmd_dispatcher; - ret = signalfd_create_sigterms(); - if (ret < 0) - goto destroy_event_loop; - server->signalfd = ret; + ret = signalfd_create_sigterms(); + if (ret < 0) + goto destroy_event_loop; + server->signalfd = ret; - ret = event_loop_add( - server->event_loop, server->signalfd, POLLIN, server_set_stopping, server); - if (ret < 0) - goto close_signalfd; + ret = event_loop_add(server->event_loop, server->signalfd, POLLIN, server_set_stopping, server); + if (ret < 0) + goto close_signalfd; - worker_queue_create(&server->worker_queue); + worker_queue_create(&server->worker_queue); - ret = storage_sqlite_settings_create(&storage_settings, settings->sqlite_path); - if (ret < 0) - goto destroy_worker_queue; + ret = storage_sqlite_settings_create(&storage_settings, settings->sqlite_path); + if (ret < 0) + goto destroy_worker_queue; - ret = storage_create(&server->storage, &storage_settings); - storage_settings_destroy(&storage_settings); - if (ret < 0) - goto destroy_worker_queue; + ret = storage_create(&server->storage, &storage_settings); + storage_settings_destroy(&storage_settings); + if (ret < 0) + goto destroy_worker_queue; - ret = storage_get_run_queue(&server->storage, &server->run_queue); - if (ret < 0) - goto destroy_storage; + ret = storage_get_run_queue(&server->storage, &server->run_queue); + if (ret < 0) + goto destroy_storage; - ret = tcp_server_create(&server->tcp_server, - server->event_loop, - settings->port, - cmd_dispatcher_handle_conn, - server->cmd_dispatcher); - if (ret < 0) - goto destroy_run_queue; + ret = tcp_server_create(&server->tcp_server, + server->event_loop, + settings->port, + cmd_dispatcher_handle_conn, + server->cmd_dispatcher); + if (ret < 0) + goto destroy_run_queue; - ret = pthread_create(&server->main_thread, NULL, server_main_thread, server); - if (ret) { - pthread_errno(ret, "pthread_create"); - goto destroy_tcp_server; - } + ret = pthread_create(&server->main_thread, NULL, server_main_thread, server); + if (ret) { + pthread_errno(ret, "pthread_create"); + goto destroy_tcp_server; + } - *_server = server; - return ret; + *_server = server; + return ret; destroy_tcp_server: - tcp_server_destroy(server->tcp_server); + tcp_server_destroy(server->tcp_server); destroy_run_queue: - run_queue_destroy(&server->run_queue); + run_queue_destroy(&server->run_queue); destroy_storage: - storage_destroy(&server->storage); + storage_destroy(&server->storage); destroy_worker_queue: - worker_queue_destroy(&server->worker_queue); + worker_queue_destroy(&server->worker_queue); close_signalfd: - signalfd_destroy(server->signalfd); + signalfd_destroy(server->signalfd); destroy_event_loop: - event_loop_destroy(server->event_loop); + event_loop_destroy(server->event_loop); destroy_cmd_dispatcher: - cmd_dispatcher_destroy(server->cmd_dispatcher); + cmd_dispatcher_destroy(server->cmd_dispatcher); destroy_cv: - pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy"); + pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy"); destroy_mtx: - pthread_errno_if(pthread_mutex_destroy(&server->server_mtx), "pthread_mutex_destroy"); + pthread_errno_if(pthread_mutex_destroy(&server->server_mtx), "pthread_mutex_destroy"); free: - free(server); + free(server); - return ret; + return ret; } void server_destroy(struct server* server) { - log("Shutting down\n"); + log("Shutting down\n"); - pthread_errno_if(pthread_join(server->main_thread, NULL), "pthread_join"); - tcp_server_destroy(server->tcp_server); - storage_destroy(&server->storage); - run_queue_destroy(&server->run_queue); - worker_queue_destroy(&server->worker_queue); - signalfd_destroy(server->signalfd); - event_loop_destroy(server->event_loop); - cmd_dispatcher_destroy(server->cmd_dispatcher); - pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy"); - pthread_errno_if(pthread_mutex_destroy(&server->server_mtx), "pthread_mutex_destroy"); - free(server); + pthread_errno_if(pthread_join(server->main_thread, NULL), "pthread_join"); + tcp_server_destroy(server->tcp_server); + storage_destroy(&server->storage); + run_queue_destroy(&server->run_queue); + worker_queue_destroy(&server->worker_queue); + signalfd_destroy(server->signalfd); + event_loop_destroy(server->event_loop); + cmd_dispatcher_destroy(server->cmd_dispatcher); + pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy"); + pthread_errno_if(pthread_mutex_destroy(&server->server_mtx), "pthread_mutex_destroy"); + free(server); } static int server_listen_thread(struct server* server) { - int ret = 0; + int ret = 0; - while (!server->stopping) { - log("Waiting for new connections\n"); + while (!server->stopping) { + log("Waiting for new connections\n"); - ret = event_loop_run(server->event_loop); - if (ret < 0) - return ret; - } + ret = event_loop_run(server->event_loop); + if (ret < 0) + return ret; + } - return 0; + return 0; } int server_main(struct server* server) { - return server_listen_thread(server); + return server_listen_thread(server); } diff --git a/src/server.h b/src/server.h index 862a752..29a9cf4 100644 --- a/src/server.h +++ b/src/server.h @@ -9,9 +9,9 @@ #define __SERVER_H__ struct settings { - const char* port; + const char* port; - const char* sqlite_path; + const char* sqlite_path; }; struct server; diff --git a/src/server_main.c b/src/server_main.c index ecbbc7e..14bc3f5 100644 --- a/src/server_main.c +++ b/src/server_main.c @@ -14,23 +14,23 @@ #include <unistd.h> static struct settings default_settings(void) { - struct settings settings = { - .port = default_port, - .sqlite_path = default_sqlite_path, - }; - return settings; + struct settings settings = { + .port = default_port, + .sqlite_path = default_sqlite_path, + }; + return settings; } const char* get_usage_string(void) { - return "[-h|--help] [-V|--version] [-v|--verbose] [-p|--port PORT] [-s|--sqlite PATH]"; + return "[-h|--help] [-V|--version] [-v|--verbose] [-p|--port PORT] [-s|--sqlite PATH]"; } static int parse_settings(struct settings* settings, int argc, char* argv[]) { - int opt, longind; + int opt, longind; - *settings = default_settings(); + *settings = default_settings(); - /* clang-format off */ + /* clang-format off */ static struct option long_options[] = { {"help", no_argument, 0, 'h'}, {"version", no_argument, 0, 'V'}, @@ -39,53 +39,53 @@ static int parse_settings(struct settings* settings, int argc, char* argv[]) { {"sqlite", required_argument, 0, 's'}, {0, 0, 0, 0}, }; - /* clang-format on */ + /* clang-format on */ - while ((opt = getopt_long(argc, argv, "hVvp:s:", long_options, &longind)) != -1) { - switch (opt) { - case 'h': - exit_with_usage(0); - break; - case 'V': - exit_with_version(); - break; - case 'v': - g_log_lvl = LOG_LVL_DEBUG; - break; - case 'p': - settings->port = optarg; - break; - case 's': - settings->sqlite_path = optarg; - break; - default: - exit_with_usage(1); - break; - } - } + while ((opt = getopt_long(argc, argv, "hVvp:s:", long_options, &longind)) != -1) { + switch (opt) { + case 'h': + exit_with_usage(0); + break; + case 'V': + exit_with_version(); + break; + case 'v': + g_log_lvl = LOG_LVL_DEBUG; + break; + case 'p': + settings->port = optarg; + break; + case 's': + settings->sqlite_path = optarg; + break; + default: + exit_with_usage(1); + break; + } + } - return 0; + return 0; } int main(int argc, char* argv[]) { - struct settings settings; - struct server* server = NULL; - int ret = 0; + struct settings settings; + struct server* server = NULL; + int ret = 0; - ret = parse_settings(&settings, argc, argv); - if (ret < 0) - return ret; + ret = parse_settings(&settings, argc, argv); + if (ret < 0) + return ret; - ret = server_create(&server, &settings); - if (ret < 0) - return ret; + ret = server_create(&server, &settings); + if (ret < 0) + return ret; - ret = server_main(server); - if (ret < 0) - goto destroy_server; + ret = server_main(server); + if (ret < 0) + goto destroy_server; destroy_server: - server_destroy(server); + server_destroy(server); - return ret; + return ret; } diff --git a/src/signal.c b/src/signal.c index 93eea31..bbfcf85 100644 --- a/src/signal.c +++ b/src/signal.c @@ -17,66 +17,66 @@ static int sigterm_signals[] = {SIGINT, SIGTERM, SIGQUIT}; static void sigterms_mask(sigset_t* set) { - sigemptyset(set); - for (size_t i = 0; i < sizeof(sigterm_signals) / sizeof(sigterm_signals[0]); ++i) - sigaddset(set, sigterm_signals[i]); + sigemptyset(set); + for (size_t i = 0; i < sizeof(sigterm_signals) / sizeof(sigterm_signals[0]); ++i) + sigaddset(set, sigterm_signals[i]); } static int signal_set_mask_internal(const sigset_t* new, sigset_t* old) { - int ret = 0; + int ret = 0; - ret = pthread_sigmask(SIG_SETMASK, new, old); - if (ret) { - pthread_errno(ret, "pthread_sigmask"); - return ret; - } + ret = pthread_sigmask(SIG_SETMASK, new, old); + if (ret) { + pthread_errno(ret, "pthread_sigmask"); + return ret; + } - return ret; + return ret; } int signal_set_mask(const sigset_t* new) { - return signal_set_mask_internal(new, NULL); + return signal_set_mask_internal(new, NULL); } int signal_block_all(sigset_t* old) { - sigset_t new; - sigfillset(&new); - return signal_set_mask_internal(&new, old); + sigset_t new; + sigfillset(&new); + return signal_set_mask_internal(&new, old); } int signal_block_sigterms(void) { - sigset_t set; - sigterms_mask(&set); - return signal_set_mask_internal(&set, NULL); + sigset_t set; + sigterms_mask(&set); + return signal_set_mask_internal(&set, NULL); } int signalfd_create(const sigset_t* set) { - static const int flags = SFD_CLOEXEC; - sigset_t old; - int ret = 0; + static const int flags = SFD_CLOEXEC; + sigset_t old; + int ret = 0; - ret = signal_set_mask_internal(set, &old); - if (ret < 0) - return ret; + ret = signal_set_mask_internal(set, &old); + if (ret < 0) + return ret; - ret = signalfd(-1, set, flags); - if (ret < 0) - goto restore; + ret = signalfd(-1, set, flags); + if (ret < 0) + goto restore; - return ret; + return ret; restore: - signal_set_mask_internal(&old, NULL); + signal_set_mask_internal(&old, NULL); - return ret; + return ret; } int signalfd_create_sigterms(void) { - sigset_t set; - sigterms_mask(&set); - return signalfd_create(&set); + sigset_t set; + sigterms_mask(&set); + return signalfd_create(&set); } void signalfd_destroy(int fd) { - file_close(fd); + file_close(fd); } diff --git a/src/sqlite.c b/src/sqlite.c index 5cb764a..11a9ce8 100644 --- a/src/sqlite.c +++ b/src/sqlite.c @@ -17,279 +17,279 @@ #include <stdlib.h> #include <string.h> -#define sqlite_errno(var, fn) \ - do { \ - log_err("%s: %s\n", fn, sqlite3_errstr(var)); \ - var = -var; \ - } while (0) +#define sqlite_errno(var, fn) \ + do { \ + log_err("%s: %s\n", fn, sqlite3_errstr(var)); \ + var = -var; \ + } while (0) -#define sqlite_errno_if(expr, fn) \ - do { \ - int CONCAT(ret, __LINE__) = expr; \ - if (CONCAT(ret, __LINE__)) \ - sqlite_errno(CONCAT(ret, __LINE__), fn); \ - } while (0) +#define sqlite_errno_if(expr, fn) \ + do { \ + int CONCAT(ret, __LINE__) = expr; \ + if (CONCAT(ret, __LINE__)) \ + sqlite_errno(CONCAT(ret, __LINE__), fn); \ + } while (0) int sqlite_init(void) { - int ret = 0; + int ret = 0; - ret = sqlite3_initialize(); - if (ret) { - sqlite_errno(ret, "sqlite3_initialize"); - return ret; - } + ret = sqlite3_initialize(); + if (ret) { + sqlite_errno(ret, "sqlite3_initialize"); + return ret; + } - return ret; + return ret; } void sqlite_destroy(void) { - sqlite_errno_if(sqlite3_shutdown(), "sqlite3_shutdown"); + sqlite_errno_if(sqlite3_shutdown(), "sqlite3_shutdown"); } static int sqlite_open(const char* path, sqlite3** db, int flags) { - int ret = 0; + int ret = 0; - ret = sqlite3_open_v2(path, db, flags, NULL); - if (ret) { - sqlite_errno(ret, "sqlite3_open_v2"); - return ret; - } + ret = sqlite3_open_v2(path, db, flags, NULL); + if (ret) { + sqlite_errno(ret, "sqlite3_open_v2"); + return ret; + } - return ret; + return ret; } int sqlite_open_rw(const char* path, sqlite3** db) { - static const int flags = SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE; - return sqlite_open(path, db, flags); + static const int flags = SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE; + return sqlite_open(path, db, flags); } int sqlite_open_ro(const char* path, sqlite3** db) { - static const int flags = SQLITE_OPEN_READONLY; - return sqlite_open(path, db, flags); + static const int flags = SQLITE_OPEN_READONLY; + return sqlite_open(path, db, flags); } void sqlite_close(sqlite3* db) { - sqlite_errno_if(sqlite3_close(db), "sqlite3_close"); + sqlite_errno_if(sqlite3_close(db), "sqlite3_close"); } int sqlite_exec(sqlite3* db, const char* stmt, sqlite3_callback callback, void* arg) { - int ret = 0; + int ret = 0; - ret = sqlite3_exec(db, stmt, callback, arg, NULL); - if (ret) { - sqlite_errno(ret, "sqlite3_exec"); - return ret; - } + ret = sqlite3_exec(db, stmt, callback, arg, NULL); + if (ret) { + sqlite_errno(ret, "sqlite3_exec"); + return ret; + } - return ret; + return ret; } int sqlite_exec_log_result(UNUSED void* arg, int numof_columns, char** values, char** column_names) { - log("Row:\n"); - for (int i = 0; i < numof_columns; ++i) { - log("\t%s: %s\n", column_names[i], values[i]); - } - return 0; + log("Row:\n"); + for (int i = 0; i < numof_columns; ++i) { + log("\t%s: %s\n", column_names[i], values[i]); + } + return 0; } int sqlite_prepare(sqlite3* db, const char* stmt, sqlite3_stmt** result) { - int ret = 0; + int ret = 0; - ret = sqlite3_prepare_v2(db, stmt, -1, result, NULL); - if (ret) { - sqlite_errno(ret, "sqlite3_prepare_v2"); - return ret; - } + ret = sqlite3_prepare_v2(db, stmt, -1, result, NULL); + if (ret) { + sqlite_errno(ret, "sqlite3_prepare_v2"); + return ret; + } - return ret; + return ret; } void sqlite_reset(sqlite3_stmt* stmt) { - sqlite_errno_if(sqlite3_reset(stmt), "sqlite3_reset"); - sqlite_errno_if(sqlite3_clear_bindings(stmt), "sqlite3_clear_bindings"); + sqlite_errno_if(sqlite3_reset(stmt), "sqlite3_reset"); + sqlite_errno_if(sqlite3_clear_bindings(stmt), "sqlite3_clear_bindings"); } void sqlite_finalize(sqlite3_stmt* stmt) { - sqlite_errno_if(sqlite3_finalize(stmt), "sqlite3_finalize"); + sqlite_errno_if(sqlite3_finalize(stmt), "sqlite3_finalize"); } int sqlite_step(sqlite3_stmt* stmt) { - int ret = 0; + int ret = 0; - ret = sqlite3_step(stmt); + ret = sqlite3_step(stmt); - switch (ret) { - case SQLITE_ROW: - return 1; - case SQLITE_DONE: - return 0; + switch (ret) { + case SQLITE_ROW: + return 1; + case SQLITE_DONE: + return 0; - default: - sqlite_errno(ret, "sqlite3_step"); - return ret; - } + default: + sqlite_errno(ret, "sqlite3_step"); + return ret; + } } int sqlite_column_int(sqlite3_stmt* stmt, int index) { - return sqlite3_column_int(stmt, index); + return sqlite3_column_int(stmt, index); } int sqlite_column_text(sqlite3_stmt* stmt, int index, char** _result) { - int ret = 0; + int ret = 0; - const unsigned char* value = sqlite3_column_text(stmt, index); - if (!value) { - ret = sqlite3_errcode(sqlite3_db_handle(stmt)); - if (ret) { - sqlite_errno(ret, "sqlite3_column_text"); - return ret; - } + const unsigned char* value = sqlite3_column_text(stmt, index); + if (!value) { + ret = sqlite3_errcode(sqlite3_db_handle(stmt)); + if (ret) { + sqlite_errno(ret, "sqlite3_column_text"); + return ret; + } - *_result = NULL; - return 0; - } + *_result = NULL; + return 0; + } - ret = sqlite3_column_bytes(stmt, index); - size_t nb = (size_t)ret; + ret = sqlite3_column_bytes(stmt, index); + size_t nb = (size_t)ret; - char* result = calloc(nb + 1, 1); - if (!result) { - log_errno("calloc"); - return -1; - } - memcpy(result, value, nb); + char* result = calloc(nb + 1, 1); + if (!result) { + log_errno("calloc"); + return -1; + } + memcpy(result, value, nb); - *_result = result; - return 0; + *_result = result; + return 0; } int sqlite_column_blob(sqlite3_stmt* stmt, int index, unsigned char** _result) { - int ret = 0; + int ret = 0; - const unsigned char* value = sqlite3_column_blob(stmt, index); - if (!value) { - ret = sqlite3_errcode(sqlite3_db_handle(stmt)); - if (ret) { - sqlite_errno(ret, "sqlite3_column_text"); - return ret; - } + const unsigned char* value = sqlite3_column_blob(stmt, index); + if (!value) { + ret = sqlite3_errcode(sqlite3_db_handle(stmt)); + if (ret) { + sqlite_errno(ret, "sqlite3_column_text"); + return ret; + } - *_result = NULL; - return 0; - } + *_result = NULL; + return 0; + } - ret = sqlite3_column_bytes(stmt, index); - size_t nb = (size_t)ret; + ret = sqlite3_column_bytes(stmt, index); + size_t nb = (size_t)ret; - unsigned char* result = malloc(nb); - if (!result) { - log_errno("malloc"); - return -1; - } - memcpy(result, value, nb); + unsigned char* result = malloc(nb); + if (!result) { + log_errno("malloc"); + return -1; + } + memcpy(result, value, nb); - *_result = result; - return 0; + *_result = result; + return 0; } int sqlite_bind_int(sqlite3_stmt* stmt, int index, int value) { - int ret = 0; + int ret = 0; - ret = sqlite3_bind_int(stmt, index, value); - if (ret) { - sqlite_errno(ret, "sqlite3_bind_int"); - return ret; - } + ret = sqlite3_bind_int(stmt, index, value); + if (ret) { + sqlite_errno(ret, "sqlite3_bind_int"); + return ret; + } - return ret; + return ret; } int sqlite_bind_text(sqlite3_stmt* stmt, int index, const char* value) { - int ret = 0; + int ret = 0; - ret = sqlite3_bind_text(stmt, index, value, -1, SQLITE_STATIC); - if (ret) { - sqlite_errno(ret, "sqlite3_bind_text"); - return ret; - } + ret = sqlite3_bind_text(stmt, index, value, -1, SQLITE_STATIC); + if (ret) { + sqlite_errno(ret, "sqlite3_bind_text"); + return ret; + } - return ret; + return ret; } int sqlite_bind_blob(sqlite3_stmt* stmt, int index, unsigned char* value, size_t nb) { - int ret = 0; + int ret = 0; - ret = sqlite3_bind_blob64(stmt, index, value, nb, SQLITE_STATIC); - if (ret) { - sqlite_errno(ret, "sqlite3_bind_blob64"); - return ret; - } + ret = sqlite3_bind_blob64(stmt, index, value, nb, SQLITE_STATIC); + if (ret) { + sqlite_errno(ret, "sqlite3_bind_blob64"); + return ret; + } - return ret; + return ret; } int sqlite_exec_as_transaction(sqlite3* db, const char* stmt) { - static const char* const fmt = "BEGIN; %s COMMIT;"; - int ret = 0; + static const char* const fmt = "BEGIN; %s COMMIT;"; + int ret = 0; - ret = snprintf(NULL, 0, fmt, stmt); - size_t nb = (size_t)ret + 1; - ret = 0; + ret = snprintf(NULL, 0, fmt, stmt); + size_t nb = (size_t)ret + 1; + ret = 0; - char* full_stmt = malloc(nb); - if (!full_stmt) { - log_errno("malloc"); - return -1; - } - snprintf(full_stmt, nb, fmt, stmt); + char* full_stmt = malloc(nb); + if (!full_stmt) { + log_errno("malloc"); + return -1; + } + snprintf(full_stmt, nb, fmt, stmt); - ret = sqlite_exec(db, stmt, NULL, NULL); - goto free; + ret = sqlite_exec(db, stmt, NULL, NULL); + goto free; free: - free(full_stmt); + free(full_stmt); - return ret; + return ret; } int sqlite_get_user_version(sqlite3* db, unsigned int* output) { - static const char* const sql = "PRAGMA user_version;"; + static const char* const sql = "PRAGMA user_version;"; - sqlite3_stmt* stmt = NULL; - int result = -1, ret = 0; + sqlite3_stmt* stmt = NULL; + int result = -1, ret = 0; - ret = sqlite_prepare(db, sql, &stmt); - if (ret < 0) - return ret; - ret = sqlite_step(stmt); - if (ret < 0) - goto finalize; - if (!ret) { - ret = -1; - log_err("Failed to read database version\n"); - goto finalize; - } + ret = sqlite_prepare(db, sql, &stmt); + if (ret < 0) + return ret; + ret = sqlite_step(stmt); + if (ret < 0) + goto finalize; + if (!ret) { + ret = -1; + log_err("Failed to read database version\n"); + goto finalize; + } - result = sqlite_column_int(stmt, 0); - if (result < 0) { - log_err("Invalid database version: %d\n", result); - ret = -1; - goto finalize; - } - *output = (unsigned int)result; + result = sqlite_column_int(stmt, 0); + if (result < 0) { + log_err("Invalid database version: %d\n", result); + ret = -1; + goto finalize; + } + *output = (unsigned int)result; - goto finalize; + goto finalize; finalize: - sqlite_finalize(stmt); + sqlite_finalize(stmt); - return ret; + return ret; } int sqlite_set_foreign_keys(sqlite3* db) { - static const char* const sql = "PRAGMA foreign_keys = ON;"; - return sqlite_exec(db, sql, NULL, NULL); + static const char* const sql = "PRAGMA foreign_keys = ON;"; + return sqlite_exec(db, sql, NULL, NULL); } diff --git a/src/storage.c b/src/storage.c index 92c0030..65df773 100644 --- a/src/storage.c +++ b/src/storage.c @@ -25,98 +25,98 @@ typedef int (*storage_get_runs_t)(struct storage*, struct run_queue*); typedef storage_get_runs_t storage_get_run_queue_t; struct storage_api { - storage_settings_destroy_t destroy_settings; - storage_create_t create; - storage_destroy_t destroy; + storage_settings_destroy_t destroy_settings; + storage_create_t create; + storage_destroy_t destroy; - storage_run_create_t run_create; - storage_run_finished_t run_finished; + storage_run_create_t run_create; + storage_run_finished_t run_finished; - storage_get_runs_t get_runs; - storage_get_run_queue_t get_run_queue; + storage_get_runs_t get_runs; + storage_get_run_queue_t get_run_queue; }; static const struct storage_api apis[] = { { - storage_sqlite_settings_destroy, - storage_sqlite_create, - storage_sqlite_destroy, + storage_sqlite_settings_destroy, + storage_sqlite_create, + storage_sqlite_destroy, - storage_sqlite_run_create, - storage_sqlite_run_finished, + storage_sqlite_run_create, + storage_sqlite_run_finished, - storage_sqlite_get_runs, - storage_sqlite_get_run_queue, + storage_sqlite_get_runs, + storage_sqlite_get_run_queue, }, }; static size_t numof_apis(void) { - return sizeof(apis) / sizeof(apis[0]); + return sizeof(apis) / sizeof(apis[0]); } static const struct storage_api* get_api(enum storage_type type) { - if (type < 0) - goto invalid_type; - if ((size_t)type > numof_apis()) - goto invalid_type; + if (type < 0) + goto invalid_type; + if ((size_t)type > numof_apis()) + goto invalid_type; - return &apis[type]; + return &apis[type]; invalid_type: - log_err("Unsupported storage type: %d\n", type); - return NULL; + log_err("Unsupported storage type: %d\n", type); + return NULL; } void storage_settings_destroy(const struct storage_settings* settings) { - const struct storage_api* api = get_api(settings->type); - if (!api) - return; - api->destroy_settings(settings); + const struct storage_api* api = get_api(settings->type); + if (!api) + return; + api->destroy_settings(settings); } int storage_create(struct storage* storage, const struct storage_settings* settings) { - int ret = 0; - const struct storage_api* api = get_api(settings->type); - if (!api) - return -1; - ret = api->create(storage, settings); - if (ret < 0) - return ret; - storage->type = settings->type; - return ret; + int ret = 0; + const struct storage_api* api = get_api(settings->type); + if (!api) + return -1; + ret = api->create(storage, settings); + if (ret < 0) + return ret; + storage->type = settings->type; + return ret; } void storage_destroy(struct storage* storage) { - const struct storage_api* api = get_api(storage->type); - if (!api) - return; - api->destroy(storage); + const struct storage_api* api = get_api(storage->type); + if (!api) + return; + api->destroy(storage); } int storage_run_create(struct storage* storage, const char* repo_url, const char* rev) { - const struct storage_api* api = get_api(storage->type); - if (!api) - return -1; - return api->run_create(storage, repo_url, rev); + const struct storage_api* api = get_api(storage->type); + if (!api) + return -1; + return api->run_create(storage, repo_url, rev); } int storage_run_finished(struct storage* storage, int run_id, const struct process_output* output) { - const struct storage_api* api = get_api(storage->type); - if (!api) - return -1; - return api->run_finished(storage, run_id, output); + const struct storage_api* api = get_api(storage->type); + if (!api) + return -1; + return api->run_finished(storage, run_id, output); } int storage_get_runs(struct storage* storage, struct run_queue* queue) { - const struct storage_api* api = get_api(storage->type); - if (!api) - return -1; - return api->get_runs(storage, queue); + const struct storage_api* api = get_api(storage->type); + if (!api) + return -1; + return api->get_runs(storage, queue); } int storage_get_run_queue(struct storage* storage, struct run_queue* queue) { - const struct storage_api* api = get_api(storage->type); - if (!api) - return -1; - return api->get_run_queue(storage, queue); + const struct storage_api* api = get_api(storage->type); + if (!api) + return -1; + return api->get_run_queue(storage, queue); } diff --git a/src/storage.h b/src/storage.h index d20e08e..e8ce59f 100644 --- a/src/storage.h +++ b/src/storage.h @@ -13,23 +13,23 @@ #include "storage_sqlite.h" enum storage_type { - STORAGE_TYPE_SQLITE, + STORAGE_TYPE_SQLITE, }; struct storage_settings { - enum storage_type type; - union { - struct storage_sqlite_settings* sqlite; - }; + enum storage_type type; + union { + struct storage_sqlite_settings* sqlite; + }; }; void storage_settings_destroy(const struct storage_settings*); struct storage { - enum storage_type type; - union { - struct storage_sqlite* sqlite; - }; + enum storage_type type; + union { + struct storage_sqlite* sqlite; + }; }; int storage_create(struct storage*, const struct storage_settings*); diff --git a/src/storage_sqlite.c b/src/storage_sqlite.c index ff651af..e8e75b5 100644 --- a/src/storage_sqlite.c +++ b/src/storage_sqlite.c @@ -22,504 +22,504 @@ #include <string.h> struct storage_sqlite_settings { - char* path; + char* path; }; int storage_sqlite_settings_create(struct storage_settings* settings, const char* path) { - struct storage_sqlite_settings* sqlite = malloc(sizeof(struct storage_sqlite_settings)); - if (!sqlite) { - log_errno("malloc"); - return -1; - } + struct storage_sqlite_settings* sqlite = malloc(sizeof(struct storage_sqlite_settings)); + if (!sqlite) { + log_errno("malloc"); + return -1; + } - sqlite->path = strdup(path); - if (!sqlite->path) { - log_errno("strdup"); - goto free; - } + sqlite->path = strdup(path); + if (!sqlite->path) { + log_errno("strdup"); + goto free; + } - settings->type = STORAGE_TYPE_SQLITE; - settings->sqlite = sqlite; - return 0; + settings->type = STORAGE_TYPE_SQLITE; + settings->sqlite = sqlite; + return 0; free: - free(sqlite); + free(sqlite); - return -1; + return -1; } void storage_sqlite_settings_destroy(const struct storage_settings* settings) { - free(settings->sqlite->path); - free(settings->sqlite); + free(settings->sqlite->path); + free(settings->sqlite); } struct prepared_stmt { - pthread_mutex_t mtx; - sqlite3_stmt* impl; + pthread_mutex_t mtx; + sqlite3_stmt* impl; }; static int prepared_stmt_init(struct prepared_stmt* stmt, sqlite3* db, const char* sql) { - int ret = 0; + int ret = 0; - ret = pthread_mutex_init(&stmt->mtx, NULL); - if (ret) { - pthread_errno(ret, "pthread_mutex_init"); - return ret; - } + ret = pthread_mutex_init(&stmt->mtx, NULL); + if (ret) { + pthread_errno(ret, "pthread_mutex_init"); + return ret; + } - ret = sqlite_prepare(db, sql, &stmt->impl); - if (ret < 0) - goto destroy_mtx; + ret = sqlite_prepare(db, sql, &stmt->impl); + if (ret < 0) + goto destroy_mtx; - return ret; + return ret; destroy_mtx: - pthread_errno_if(pthread_mutex_destroy(&stmt->mtx), "pthread_mutex_destroy"); + pthread_errno_if(pthread_mutex_destroy(&stmt->mtx), "pthread_mutex_destroy"); - return ret; + return ret; } static void prepared_stmt_destroy(struct prepared_stmt* stmt) { - pthread_errno_if(pthread_mutex_destroy(&stmt->mtx), "pthread_mutex_destroy"); - sqlite_finalize(stmt->impl); + pthread_errno_if(pthread_mutex_destroy(&stmt->mtx), "pthread_mutex_destroy"); + sqlite_finalize(stmt->impl); } static int prepared_stmt_lock(struct prepared_stmt* stmt) { - int ret = pthread_mutex_lock(&stmt->mtx); - if (ret) { - pthread_errno(ret, "pthread_mutex_unlock"); - return ret; - } - return ret; + int ret = pthread_mutex_lock(&stmt->mtx); + if (ret) { + pthread_errno(ret, "pthread_mutex_unlock"); + return ret; + } + return ret; } static void prepared_stmt_unlock(struct prepared_stmt* stmt) { - pthread_errno_if(pthread_mutex_unlock(&stmt->mtx), "pthread_mutex_unlock"); + pthread_errno_if(pthread_mutex_unlock(&stmt->mtx), "pthread_mutex_unlock"); } struct storage_sqlite { - sqlite3* db; + sqlite3* db; - struct prepared_stmt stmt_repo_find; - struct prepared_stmt stmt_repo_insert; - struct prepared_stmt stmt_run_insert; - struct prepared_stmt stmt_run_finished; - struct prepared_stmt stmt_get_runs; - struct prepared_stmt stmt_get_run_queue; + struct prepared_stmt stmt_repo_find; + struct prepared_stmt stmt_repo_insert; + struct prepared_stmt stmt_run_insert; + struct prepared_stmt stmt_run_finished; + struct prepared_stmt stmt_get_runs; + struct prepared_stmt stmt_get_run_queue; }; static int storage_sqlite_upgrade_to(struct storage_sqlite* storage, size_t version) { - static const char* const fmt = "%s PRAGMA user_version = %zu;"; + static const char* const fmt = "%s PRAGMA user_version = %zu;"; - const char* script = sqlite_schemas[version]; - int ret = 0; + const char* script = sqlite_schemas[version]; + int ret = 0; - ret = snprintf(NULL, 0, fmt, script, version + 1); - size_t nb = (size_t)ret + 1; - ret = 0; + ret = snprintf(NULL, 0, fmt, script, version + 1); + size_t nb = (size_t)ret + 1; + ret = 0; - char* full_script = malloc(nb); - if (!full_script) { - log_errno("malloc"); - return -1; - } - snprintf(full_script, nb, fmt, script, version + 1); + char* full_script = malloc(nb); + if (!full_script) { + log_errno("malloc"); + return -1; + } + snprintf(full_script, nb, fmt, script, version + 1); - ret = sqlite_exec_as_transaction(storage->db, full_script); - goto free; + ret = sqlite_exec_as_transaction(storage->db, full_script); + goto free; free: - free(full_script); + free(full_script); - return ret; + return ret; } static int storage_sqlite_upgrade_from_to(struct storage_sqlite* storage, size_t from, size_t to) { - int ret = 0; + int ret = 0; - for (size_t i = from; i < to; ++i) { - log("Upgrading SQLite database from version %zu to version %zu\n", i, i + 1); - ret = storage_sqlite_upgrade_to(storage, i); - if (ret < 0) { - log_err("Failed to upgrade to version %zu\n", i + 1); - return ret; - } - } + for (size_t i = from; i < to; ++i) { + log("Upgrading SQLite database from version %zu to version %zu\n", i, i + 1); + ret = storage_sqlite_upgrade_to(storage, i); + if (ret < 0) { + log_err("Failed to upgrade to version %zu\n", i + 1); + return ret; + } + } - return ret; + return ret; } static int storage_sqlite_upgrade(struct storage_sqlite* storage) { - unsigned int current_version = 0; - int ret = 0; + unsigned int current_version = 0; + int ret = 0; - ret = sqlite_get_user_version(storage->db, ¤t_version); - if (ret < 0) - return ret; - log("SQLite database version: %u\n", current_version); + ret = sqlite_get_user_version(storage->db, ¤t_version); + if (ret < 0) + return ret; + log("SQLite database version: %u\n", current_version); - size_t newest_version = sizeof(sqlite_schemas) / sizeof(sqlite_schemas[0]); - log("Newest database version: %zu\n", newest_version); + size_t newest_version = sizeof(sqlite_schemas) / sizeof(sqlite_schemas[0]); + log("Newest database version: %zu\n", newest_version); - if (current_version > newest_version) { - log_err("Unknown database version: %u\n", current_version); - return -1; - } + if (current_version > newest_version) { + log_err("Unknown database version: %u\n", current_version); + return -1; + } - if (current_version == newest_version) { - log("SQLite database already at the newest version\n"); - return 0; - } + if (current_version == newest_version) { + log("SQLite database already at the newest version\n"); + return 0; + } - return storage_sqlite_upgrade_from_to(storage, current_version, newest_version); + return storage_sqlite_upgrade_from_to(storage, current_version, newest_version); } static int storage_sqlite_setup(struct storage_sqlite* storage) { - int ret = 0; + int ret = 0; - ret = sqlite_set_foreign_keys(storage->db); - if (ret < 0) - return ret; + ret = sqlite_set_foreign_keys(storage->db); + if (ret < 0) + return ret; - ret = storage_sqlite_upgrade(storage); - if (ret < 0) - return ret; + ret = storage_sqlite_upgrade(storage); + if (ret < 0) + return ret; - return ret; + return ret; } static int storage_sqlite_prepare_statements(struct storage_sqlite* storage) { - static const char* const fmt_repo_find = "SELECT id FROM cimple_repos WHERE url = ?;"; - static const char* const fmt_repo_insert = - "INSERT INTO cimple_repos(url) VALUES (?) ON CONFLICT(url) DO NOTHING;"; - static const char* const fmt_run_insert = - "INSERT INTO cimple_runs(status, exit_code, output, repo_id, repo_rev) VALUES (?, -1, x'', ?, ?) RETURNING id;"; - static const char* const fmt_run_finished = - "UPDATE cimple_runs SET status = ?, exit_code = ?, output = ? WHERE id = ?;"; - static const char* const fmt_get_runs = - "SELECT id, status, exit_code, repo_url, repo_rev FROM cimple_runs_view ORDER BY id DESC"; - static const char* const fmt_get_run_queue = - "SELECT id, status, exit_code, repo_url, repo_rev FROM cimple_runs_view WHERE status = ? ORDER BY id;"; + static const char* const fmt_repo_find = "SELECT id FROM cimple_repos WHERE url = ?;"; + static const char* const fmt_repo_insert = + "INSERT INTO cimple_repos(url) VALUES (?) ON CONFLICT(url) DO NOTHING;"; + static const char* const fmt_run_insert = + "INSERT INTO cimple_runs(status, exit_code, output, repo_id, repo_rev) VALUES (?, -1, x'', ?, ?) RETURNING id;"; + static const char* const fmt_run_finished = + "UPDATE cimple_runs SET status = ?, exit_code = ?, output = ? WHERE id = ?;"; + static const char* const fmt_get_runs = + "SELECT id, status, exit_code, repo_url, repo_rev FROM cimple_runs_view ORDER BY id DESC"; + static const char* const fmt_get_run_queue = + "SELECT id, status, exit_code, repo_url, repo_rev FROM cimple_runs_view WHERE status = ? ORDER BY id;"; - int ret = 0; + int ret = 0; - ret = prepared_stmt_init(&storage->stmt_repo_find, storage->db, fmt_repo_find); - if (ret < 0) - return ret; - ret = prepared_stmt_init(&storage->stmt_repo_insert, storage->db, fmt_repo_insert); - if (ret < 0) - goto finalize_repo_find; - ret = prepared_stmt_init(&storage->stmt_run_insert, storage->db, fmt_run_insert); - if (ret < 0) - goto finalize_repo_insert; - ret = prepared_stmt_init(&storage->stmt_run_finished, storage->db, fmt_run_finished); - if (ret < 0) - goto finalize_run_insert; - ret = prepared_stmt_init(&storage->stmt_get_runs, storage->db, fmt_get_runs); - if (ret < 0) - goto finalize_run_finished; - ret = prepared_stmt_init(&storage->stmt_get_run_queue, storage->db, fmt_get_run_queue); - if (ret < 0) - goto finalize_get_runs; + ret = prepared_stmt_init(&storage->stmt_repo_find, storage->db, fmt_repo_find); + if (ret < 0) + return ret; + ret = prepared_stmt_init(&storage->stmt_repo_insert, storage->db, fmt_repo_insert); + if (ret < 0) + goto finalize_repo_find; + ret = prepared_stmt_init(&storage->stmt_run_insert, storage->db, fmt_run_insert); + if (ret < 0) + goto finalize_repo_insert; + ret = prepared_stmt_init(&storage->stmt_run_finished, storage->db, fmt_run_finished); + if (ret < 0) + goto finalize_run_insert; + ret = prepared_stmt_init(&storage->stmt_get_runs, storage->db, fmt_get_runs); + if (ret < 0) + goto finalize_run_finished; + ret = prepared_stmt_init(&storage->stmt_get_run_queue, storage->db, fmt_get_run_queue); + if (ret < 0) + goto finalize_get_runs; - return ret; + return ret; finalize_get_runs: - prepared_stmt_destroy(&storage->stmt_get_runs); + prepared_stmt_destroy(&storage->stmt_get_runs); finalize_run_finished: - prepared_stmt_destroy(&storage->stmt_run_finished); + prepared_stmt_destroy(&storage->stmt_run_finished); finalize_run_insert: - prepared_stmt_destroy(&storage->stmt_run_insert); + prepared_stmt_destroy(&storage->stmt_run_insert); finalize_repo_insert: - prepared_stmt_destroy(&storage->stmt_repo_insert); + prepared_stmt_destroy(&storage->stmt_repo_insert); finalize_repo_find: - prepared_stmt_destroy(&storage->stmt_repo_find); + prepared_stmt_destroy(&storage->stmt_repo_find); - return ret; + return ret; } static void storage_sqlite_finalize_statements(struct storage_sqlite* storage) { - prepared_stmt_destroy(&storage->stmt_get_run_queue); - prepared_stmt_destroy(&storage->stmt_get_runs); - prepared_stmt_destroy(&storage->stmt_run_finished); - prepared_stmt_destroy(&storage->stmt_run_insert); - prepared_stmt_destroy(&storage->stmt_repo_insert); - prepared_stmt_destroy(&storage->stmt_repo_find); + prepared_stmt_destroy(&storage->stmt_get_run_queue); + prepared_stmt_destroy(&storage->stmt_get_runs); + prepared_stmt_destroy(&storage->stmt_run_finished); + prepared_stmt_destroy(&storage->stmt_run_insert); + prepared_stmt_destroy(&storage->stmt_repo_insert); + prepared_stmt_destroy(&storage->stmt_repo_find); } int storage_sqlite_create(struct storage* storage, const struct storage_settings* settings) { - int ret = 0; + int ret = 0; - log("Using SQLite database at %s\n", settings->sqlite->path); + log("Using SQLite database at %s\n", settings->sqlite->path); - struct storage_sqlite* sqlite = malloc(sizeof(struct storage_sqlite)); - if (!sqlite) { - log_errno("malloc"); - return -1; - } + struct storage_sqlite* sqlite = malloc(sizeof(struct storage_sqlite)); + if (!sqlite) { + log_errno("malloc"); + return -1; + } - ret = sqlite_init(); - if (ret < 0) - goto free; - ret = sqlite_open_rw(settings->sqlite->path, &sqlite->db); - if (ret < 0) - goto destroy; - ret = storage_sqlite_setup(sqlite); - if (ret < 0) - goto close; - ret = storage_sqlite_prepare_statements(sqlite); - if (ret < 0) - goto close; + ret = sqlite_init(); + if (ret < 0) + goto free; + ret = sqlite_open_rw(settings->sqlite->path, &sqlite->db); + if (ret < 0) + goto destroy; + ret = storage_sqlite_setup(sqlite); + if (ret < 0) + goto close; + ret = storage_sqlite_prepare_statements(sqlite); + if (ret < 0) + goto close; - storage->sqlite = sqlite; - return ret; + storage->sqlite = sqlite; + return ret; close: - sqlite_close(storage->sqlite->db); + sqlite_close(storage->sqlite->db); destroy: - sqlite_destroy(); + sqlite_destroy(); free: - free(sqlite); + free(sqlite); - return ret; + return ret; } void storage_sqlite_destroy(struct storage* storage) { - storage_sqlite_finalize_statements(storage->sqlite); - sqlite_close(storage->sqlite->db); - sqlite_destroy(); - free(storage->sqlite); + storage_sqlite_finalize_statements(storage->sqlite); + sqlite_close(storage->sqlite->db); + sqlite_destroy(); + free(storage->sqlite); } static int storage_sqlite_find_repo(struct storage_sqlite* storage, const char* url) { - struct prepared_stmt* stmt = &storage->stmt_repo_find; - int ret = 0; + struct prepared_stmt* stmt = &storage->stmt_repo_find; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = sqlite_bind_text(stmt->impl, 1, url); - if (ret < 0) - goto reset; - ret = sqlite_step(stmt->impl); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = sqlite_bind_text(stmt->impl, 1, url); + if (ret < 0) + goto reset; + ret = sqlite_step(stmt->impl); + if (ret < 0) + goto reset; - if (!ret) - goto reset; + if (!ret) + goto reset; - ret = sqlite_column_int(stmt->impl, 0); - goto reset; + ret = sqlite_column_int(stmt->impl, 0); + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - return ret; + return ret; } static int storage_sqlite_insert_repo(struct storage_sqlite* storage, const char* url) { - struct prepared_stmt* stmt = &storage->stmt_repo_insert; - int ret = 0; + struct prepared_stmt* stmt = &storage->stmt_repo_insert; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = sqlite_bind_text(stmt->impl, 1, url); - if (ret < 0) - goto reset; - ret = sqlite_step(stmt->impl); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = sqlite_bind_text(stmt->impl, 1, url); + if (ret < 0) + goto reset; + ret = sqlite_step(stmt->impl); + if (ret < 0) + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - if (ret < 0) - return ret; + if (ret < 0) + return ret; - return storage_sqlite_find_repo(storage, url); + return storage_sqlite_find_repo(storage, url); } static int storage_sqlite_insert_run(struct storage_sqlite* storage, int repo_id, const char* rev) { - struct prepared_stmt* stmt = &storage->stmt_run_insert; - int ret = 0; + struct prepared_stmt* stmt = &storage->stmt_run_insert; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_CREATED); - if (ret < 0) - goto reset; - ret = sqlite_bind_int(stmt->impl, 2, repo_id); - if (ret < 0) - goto reset; - ret = sqlite_bind_text(stmt->impl, 3, rev); - if (ret < 0) - goto reset; - ret = sqlite_step(stmt->impl); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_CREATED); + if (ret < 0) + goto reset; + ret = sqlite_bind_int(stmt->impl, 2, repo_id); + if (ret < 0) + goto reset; + ret = sqlite_bind_text(stmt->impl, 3, rev); + if (ret < 0) + goto reset; + ret = sqlite_step(stmt->impl); + if (ret < 0) + goto reset; - if (!ret) { - ret = -1; - log_err("Failed to insert a run\n"); - goto reset; - } + if (!ret) { + ret = -1; + log_err("Failed to insert a run\n"); + goto reset; + } - ret = sqlite_column_int(stmt->impl, 0); - goto reset; + ret = sqlite_column_int(stmt->impl, 0); + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - return ret; + return ret; } int storage_sqlite_run_create(struct storage* storage, const char* repo_url, const char* rev) { - int ret = 0; + int ret = 0; - ret = storage_sqlite_insert_repo(storage->sqlite, repo_url); - if (ret < 0) - return ret; + ret = storage_sqlite_insert_repo(storage->sqlite, repo_url); + if (ret < 0) + return ret; - ret = storage_sqlite_insert_run(storage->sqlite, ret, rev); - if (ret < 0) - return ret; + ret = storage_sqlite_insert_run(storage->sqlite, ret, rev); + if (ret < 0) + return ret; - return ret; + return ret; } int storage_sqlite_run_finished(struct storage* storage, int run_id, const struct process_output* output) { - struct prepared_stmt* stmt = &storage->sqlite->stmt_run_finished; - int ret = 0; + struct prepared_stmt* stmt = &storage->sqlite->stmt_run_finished; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_FINISHED); - if (ret < 0) - goto reset; - ret = sqlite_bind_int(stmt->impl, 2, output->ec); - if (ret < 0) - goto reset; - ret = sqlite_bind_blob(stmt->impl, 3, output->data, output->data_size); - if (ret < 0) - goto reset; - ret = sqlite_bind_int(stmt->impl, 4, run_id); - if (ret < 0) - goto reset; - ret = sqlite_step(stmt->impl); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_FINISHED); + if (ret < 0) + goto reset; + ret = sqlite_bind_int(stmt->impl, 2, output->ec); + if (ret < 0) + goto reset; + ret = sqlite_bind_blob(stmt->impl, 3, output->data, output->data_size); + if (ret < 0) + goto reset; + ret = sqlite_bind_int(stmt->impl, 4, run_id); + if (ret < 0) + goto reset; + ret = sqlite_step(stmt->impl); + if (ret < 0) + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - return ret; + return ret; } static int storage_sqlite_row_to_run(struct sqlite3_stmt* stmt, struct run** run) { - int ret = 0; + int ret = 0; - int id = sqlite_column_int(stmt, 0); - int status = sqlite_column_int(stmt, 1); - int exit_code = sqlite_column_int(stmt, 2); + int id = sqlite_column_int(stmt, 0); + int status = sqlite_column_int(stmt, 1); + int exit_code = sqlite_column_int(stmt, 2); - char* url = NULL; - ret = sqlite_column_text(stmt, 3, &url); - if (ret < 0) - return ret; + char* url = NULL; + ret = sqlite_column_text(stmt, 3, &url); + if (ret < 0) + return ret; - char* rev = NULL; - ret = sqlite_column_text(stmt, 4, &rev); - if (ret < 0) - goto free_url; + char* rev = NULL; + ret = sqlite_column_text(stmt, 4, &rev); + if (ret < 0) + goto free_url; - ret = run_new(run, id, url, rev, status, exit_code); - if (ret < 0) - goto free_rev; + ret = run_new(run, id, url, rev, status, exit_code); + if (ret < 0) + goto free_rev; - log("Adding run %d for repository %s to the queue\n", id, url); + log("Adding run %d for repository %s to the queue\n", id, url); free_rev: - free(rev); + free(rev); free_url: - free(url); + free(url); - return ret; + return ret; } static int storage_sqlite_rows_to_runs(struct sqlite3_stmt* stmt, struct run_queue* queue) { - int ret = 0; + int ret = 0; - run_queue_create(queue); + run_queue_create(queue); - while (1) { - ret = sqlite_step(stmt); - if (!ret) - break; - if (ret < 0) - goto run_queue_destroy; + while (1) { + ret = sqlite_step(stmt); + if (!ret) + break; + if (ret < 0) + goto run_queue_destroy; - struct run* run = NULL; + struct run* run = NULL; - ret = storage_sqlite_row_to_run(stmt, &run); - if (ret < 0) - goto run_queue_destroy; + ret = storage_sqlite_row_to_run(stmt, &run); + if (ret < 0) + goto run_queue_destroy; - run_queue_add_last(queue, run); - } + run_queue_add_last(queue, run); + } - return ret; + return ret; run_queue_destroy: - run_queue_destroy(queue); + run_queue_destroy(queue); - return ret; + return ret; } int storage_sqlite_get_runs(struct storage* storage, struct run_queue* queue) { - struct prepared_stmt* stmt = &storage->sqlite->stmt_get_runs; - int ret = 0; + struct prepared_stmt* stmt = &storage->sqlite->stmt_get_runs; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = storage_sqlite_rows_to_runs(stmt->impl, queue); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = storage_sqlite_rows_to_runs(stmt->impl, queue); + if (ret < 0) + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - return ret; + return ret; } int storage_sqlite_get_run_queue(struct storage* storage, struct run_queue* queue) { - struct prepared_stmt* stmt = &storage->sqlite->stmt_get_run_queue; - int ret = 0; + struct prepared_stmt* stmt = &storage->sqlite->stmt_get_run_queue; + int ret = 0; - ret = prepared_stmt_lock(stmt); - if (ret < 0) - return ret; - ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_CREATED); - if (ret < 0) - goto reset; - ret = storage_sqlite_rows_to_runs(stmt->impl, queue); - if (ret < 0) - goto reset; + ret = prepared_stmt_lock(stmt); + if (ret < 0) + return ret; + ret = sqlite_bind_int(stmt->impl, 1, RUN_STATUS_CREATED); + if (ret < 0) + goto reset; + ret = storage_sqlite_rows_to_runs(stmt->impl, queue); + if (ret < 0) + goto reset; reset: - sqlite_reset(stmt->impl); - prepared_stmt_unlock(stmt); + sqlite_reset(stmt->impl); + prepared_stmt_unlock(stmt); - return ret; + return ret; } diff --git a/src/string.c b/src/string.c index 7c09f4f..1d2c3f2 100644 --- a/src/string.c +++ b/src/string.c @@ -16,35 +16,35 @@ /* glibc calls this stpecpy; it's not provided by glibc; however, it does * provide a possible implementation in string_copying(7), which I copied from. */ char* string_append(char* dst, char* end, const char* src) { - if (!dst) - return NULL; - if (dst == end) - return end; + if (!dst) + return NULL; + if (dst == end) + return end; - char* p = memccpy(dst, src, '\0', end - dst); - if (p) - return p - 1; + char* p = memccpy(dst, src, '\0', end - dst); + if (p) + return p - 1; - end[-1] = '\0'; - return end; + end[-1] = '\0'; + return end; } int string_to_int(const char* src, int* result) { - char* endptr = NULL; + char* endptr = NULL; - errno = 0; - long ret = strtol(src, &endptr, 10); + errno = 0; + long ret = strtol(src, &endptr, 10); - if (errno) { - log_errno("strtol"); - return -1; - } + if (errno) { + log_errno("strtol"); + return -1; + } - if (endptr == src || *endptr != '\0') { - log_err("Invalid number: %s\n", src); - return -1; - } + if (endptr == src || *endptr != '\0') { + log_err("Invalid number: %s\n", src); + return -1; + } - *result = (int)ret; - return 0; + *result = (int)ret; + return 0; } diff --git a/src/tcp_server.c b/src/tcp_server.c index f5c39fc..cdd5db6 100644 --- a/src/tcp_server.c +++ b/src/tcp_server.c @@ -60,161 +60,161 @@ */ struct client { - struct tcp_server* server; - int conn_fd; + struct tcp_server* server; + int conn_fd; - int cleanup_fd; + int cleanup_fd; - pid_t tid; - pthread_t thread; + pid_t tid; + pthread_t thread; - SIMPLEQ_ENTRY(client) entries; + SIMPLEQ_ENTRY(client) entries; }; SIMPLEQ_HEAD(client_queue, client); struct tcp_server { - struct event_loop* loop; + struct event_loop* loop; - tcp_server_conn_handler conn_handler; - void* conn_handler_arg; + tcp_server_conn_handler conn_handler; + void* conn_handler_arg; - struct client_queue client_queue; + struct client_queue client_queue; - int accept_fd; + int accept_fd; }; static void client_destroy(struct client* client) { - log_debug("Cleaning up client thread %d\n", client->tid); + log_debug("Cleaning up client thread %d\n", client->tid); - SIMPLEQ_REMOVE(&client->server->client_queue, client, client, entries); - pthread_errno_if(pthread_join(client->thread, NULL), "pthread_join"); - file_close(client->cleanup_fd); - net_close(client->conn_fd); - free(client); + SIMPLEQ_REMOVE(&client->server->client_queue, client, client, entries); + pthread_errno_if(pthread_join(client->thread, NULL), "pthread_join"); + file_close(client->cleanup_fd); + net_close(client->conn_fd); + free(client); } static int client_destroy_handler(UNUSED struct event_loop* loop, UNUSED int fd, UNUSED short revents, void* _client) { - struct client* client = (struct client*)_client; - log_debug("Client thread %d indicated that it's done\n", client->tid); + struct client* client = (struct client*)_client; + log_debug("Client thread %d indicated that it's done\n", client->tid); - client_destroy(client); - return 0; + client_destroy(client); + return 0; } static void* client_thread_func(void* _client) { - struct client* client = (struct client*)_client; - int ret = 0; + struct client* client = (struct client*)_client; + int ret = 0; - client->tid = gettid(); - log_debug("New client thread thread %d has started\n", client->tid); + client->tid = gettid(); + log_debug("New client thread thread %d has started\n", client->tid); - /* Let the client thread handle its signals except those that should be - * handled in the main thread. */ - ret = signal_block_sigterms(); - if (ret < 0) - goto cleanup; + /* Let the client thread handle its signals except those that should be + * handled in the main thread. */ + ret = signal_block_sigterms(); + if (ret < 0) + goto cleanup; - ret = client->server->conn_handler(client->conn_fd, client->server->conn_handler_arg); - if (ret < 0) - goto cleanup; + ret = client->server->conn_handler(client->conn_fd, client->server->conn_handler_arg); + if (ret < 0) + goto cleanup; cleanup: - log_errno_if(eventfd_write(client->cleanup_fd, 1), "eventfd_write"); + log_errno_if(eventfd_write(client->cleanup_fd, 1), "eventfd_write"); - return NULL; + return NULL; } static int client_create_thread(struct client* client) { - sigset_t old_mask; - int ret = 0; + sigset_t old_mask; + int ret = 0; - /* Block all signals (we'll unblock them later); the client thread will - * have all signals blocked initially. This allows the main thread to - * handle SIGINT/SIGTERM/etc. */ - ret = signal_block_all(&old_mask); - if (ret < 0) - return ret; + /* Block all signals (we'll unblock them later); the client thread will + * have all signals blocked initially. This allows the main thread to + * handle SIGINT/SIGTERM/etc. */ + ret = signal_block_all(&old_mask); + if (ret < 0) + return ret; - ret = pthread_create(&client->thread, NULL, client_thread_func, client); - if (ret) { - pthread_errno(ret, "pthread_create"); - goto restore_mask; - } + ret = pthread_create(&client->thread, NULL, client_thread_func, client); + if (ret) { + pthread_errno(ret, "pthread_create"); + goto restore_mask; + } restore_mask: - /* Restore the previously-enabled signals for handling in the main thread. */ - signal_set_mask(&old_mask); + /* Restore the previously-enabled signals for handling in the main thread. */ + signal_set_mask(&old_mask); - return ret; + return ret; } static int client_create(struct tcp_server* server, int conn_fd) { - int ret = 0; + int ret = 0; - struct client* client = calloc(1, sizeof(struct client)); - if (!client) { - log_errno("calloc"); - return -1; - } + struct client* client = calloc(1, sizeof(struct client)); + if (!client) { + log_errno("calloc"); + return -1; + } - client->server = server; - client->conn_fd = conn_fd; + client->server = server; + client->conn_fd = conn_fd; - ret = eventfd(0, EFD_CLOEXEC); - if (ret < 0) { - log_errno("eventfd"); - goto free; - } - client->cleanup_fd = ret; + ret = eventfd(0, EFD_CLOEXEC); + if (ret < 0) { + log_errno("eventfd"); + goto free; + } + client->cleanup_fd = ret; - ret = event_loop_add_once( - server->loop, client->cleanup_fd, POLLIN, client_destroy_handler, client); - if (ret < 0) - goto close_cleanup_fd; + ret = event_loop_add_once( + server->loop, client->cleanup_fd, POLLIN, client_destroy_handler, client); + if (ret < 0) + goto close_cleanup_fd; - SIMPLEQ_INSERT_TAIL(&server->client_queue, client, entries); + SIMPLEQ_INSERT_TAIL(&server->client_queue, client, entries); - ret = client_create_thread(client); - if (ret < 0) - goto remove_from_client_queue; + ret = client_create_thread(client); + if (ret < 0) + goto remove_from_client_queue; - return ret; + return ret; remove_from_client_queue: - SIMPLEQ_REMOVE(&server->client_queue, client, client, entries); + SIMPLEQ_REMOVE(&server->client_queue, client, client, entries); close_cleanup_fd: - file_close(client->cleanup_fd); + file_close(client->cleanup_fd); free: - free(client); + free(client); - return ret; + return ret; } static void client_queue_create(struct client_queue* client_queue) { - SIMPLEQ_INIT(client_queue); + SIMPLEQ_INIT(client_queue); } static void client_queue_destroy(struct client_queue* client_queue) { - struct client* entry1 = SIMPLEQ_FIRST(client_queue); - while (entry1) { - struct client* entry2 = SIMPLEQ_NEXT(entry1, entries); - client_destroy(entry1); - entry1 = entry2; - } + struct client* entry1 = SIMPLEQ_FIRST(client_queue); + while (entry1) { + struct client* entry2 = SIMPLEQ_NEXT(entry1, entries); + client_destroy(entry1); + entry1 = entry2; + } } static int tcp_server_accept_handler(UNUSED struct event_loop* loop, UNUSED int fd, UNUSED short revents, void* _server) { - struct tcp_server* server = (struct tcp_server*)_server; - return tcp_server_accept(server); + struct tcp_server* server = (struct tcp_server*)_server; + return tcp_server_accept(server); } int tcp_server_create(struct tcp_server** _server, @@ -222,63 +222,63 @@ int tcp_server_create(struct tcp_server** _server, const char* port, tcp_server_conn_handler conn_handler, void* conn_handler_arg) { - int ret = 0; + int ret = 0; - struct tcp_server* server = calloc(1, sizeof(struct tcp_server)); - if (!server) { - log_errno("calloc"); - return -1; - } + struct tcp_server* server = calloc(1, sizeof(struct tcp_server)); + if (!server) { + log_errno("calloc"); + return -1; + } - server->loop = loop; + server->loop = loop; - server->conn_handler = conn_handler; - server->conn_handler_arg = conn_handler_arg; + server->conn_handler = conn_handler; + server->conn_handler_arg = conn_handler_arg; - client_queue_create(&server->client_queue); + client_queue_create(&server->client_queue); - ret = net_bind(port); - if (ret < 0) - goto free; - server->accept_fd = ret; + ret = net_bind(port); + if (ret < 0) + goto free; + server->accept_fd = ret; - ret = event_loop_add(loop, server->accept_fd, POLLIN, tcp_server_accept_handler, server); - if (ret < 0) - goto close; + ret = event_loop_add(loop, server->accept_fd, POLLIN, tcp_server_accept_handler, server); + if (ret < 0) + goto close; - *_server = server; - return ret; + *_server = server; + return ret; close: - net_close(server->accept_fd); + net_close(server->accept_fd); free: - free(server); + free(server); - return ret; + return ret; } void tcp_server_destroy(struct tcp_server* server) { - net_close(server->accept_fd); - client_queue_destroy(&server->client_queue); - free(server); + net_close(server->accept_fd); + client_queue_destroy(&server->client_queue); + free(server); } int tcp_server_accept(struct tcp_server* server) { - int conn_fd = -1, ret = 0; + int conn_fd = -1, ret = 0; - ret = net_accept(server->accept_fd); - if (ret < 0) - return ret; - conn_fd = ret; + ret = net_accept(server->accept_fd); + if (ret < 0) + return ret; + conn_fd = ret; - ret = client_create(server, conn_fd); - if (ret < 0) - goto close_conn; + ret = client_create(server, conn_fd); + if (ret < 0) + goto close_conn; - return ret; + return ret; close_conn: - net_close(conn_fd); + net_close(conn_fd); - return ret; + return ret; } diff --git a/src/worker.c b/src/worker.c index b64152c..dd271a5 100644 --- a/src/worker.c +++ b/src/worker.c @@ -25,75 +25,75 @@ #include <string.h> struct worker { - struct settings* settings; + struct settings* settings; - int stopping; + int stopping; - struct cmd_dispatcher* cmd_dispatcher; + struct cmd_dispatcher* cmd_dispatcher; - struct event_loop* event_loop; - int signalfd; + struct event_loop* event_loop; + int signalfd; - struct run* run; + struct run* run; }; static struct settings* worker_settings_copy(const struct settings* src) { - struct settings* result = malloc(sizeof(*src)); - if (!result) { - log_errno("malloc"); - return NULL; - } + struct settings* result = malloc(sizeof(*src)); + if (!result) { + log_errno("malloc"); + return NULL; + } - result->host = strdup(src->host); - if (!result->host) { - log_errno("strdup"); - goto free_result; - } + result->host = strdup(src->host); + if (!result->host) { + log_errno("strdup"); + goto free_result; + } - result->port = strdup(src->port); - if (!result->port) { - log_errno("strdup"); - goto free_host; - } + result->port = strdup(src->port); + if (!result->port) { + log_errno("strdup"); + goto free_host; + } - return result; + return result; free_host: - free((void*)result->host); + free((void*)result->host); free_result: - free(result); + free(result); - return NULL; + return NULL; } static void worker_settings_destroy(struct settings* settings) { - free((void*)settings->port); - free((void*)settings->host); - free(settings); + free((void*)settings->port); + free((void*)settings->host); + free(settings); } static int worker_set_stopping(UNUSED struct event_loop* loop, UNUSED int fd, UNUSED short revents, void* _worker) { - struct worker* worker = (struct worker*)_worker; - worker->stopping = 1; - return 0; + struct worker* worker = (struct worker*)_worker; + worker->stopping = 1; + return 0; } static int worker_handle_cmd_start_run(const struct jsonrpc_request* request, UNUSED struct jsonrpc_response** response, void* _ctx) { - struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; - struct worker* worker = (struct worker*)ctx->arg; - int ret = 0; + struct cmd_conn_ctx* ctx = (struct cmd_conn_ctx*)_ctx; + struct worker* worker = (struct worker*)ctx->arg; + int ret = 0; - ret = request_parse_start_run(request, &worker->run); - if (ret < 0) - return ret; + ret = request_parse_start_run(request, &worker->run); + if (ret < 0) + return ret; - return ret; + return ret; } static struct cmd_desc commands[] = { @@ -103,171 +103,170 @@ static struct cmd_desc commands[] = { static const size_t numof_commands = sizeof(commands) / sizeof(commands[0]); int worker_create(struct worker** _worker, const struct settings* settings) { - int ret = 0; + int ret = 0; - struct worker* worker = malloc(sizeof(struct worker)); - if (!worker) { - log_errno("malloc"); - return -1; - } + struct worker* worker = malloc(sizeof(struct worker)); + if (!worker) { + log_errno("malloc"); + return -1; + } - worker->settings = worker_settings_copy(settings); - if (!worker->settings) { - ret = -1; - goto free; - } + worker->settings = worker_settings_copy(settings); + if (!worker->settings) { + ret = -1; + goto free; + } - worker->stopping = 0; + worker->stopping = 0; - ret = cmd_dispatcher_create(&worker->cmd_dispatcher, commands, numof_commands, worker); - if (ret < 0) - goto free_settings; + ret = cmd_dispatcher_create(&worker->cmd_dispatcher, commands, numof_commands, worker); + if (ret < 0) + goto free_settings; - ret = event_loop_create(&worker->event_loop); - if (ret < 0) - goto destroy_cmd_dispatcher; + ret = event_loop_create(&worker->event_loop); + if (ret < 0) + goto destroy_cmd_dispatcher; - ret = signalfd_create_sigterms(); - if (ret < 0) - goto destroy_event_loop; - worker->signalfd = ret; + ret = signalfd_create_sigterms(); + if (ret < 0) + goto destroy_event_loop; + worker->signalfd = ret; - ret = event_loop_add( - worker->event_loop, worker->signalfd, POLLIN, worker_set_stopping, worker); - if (ret < 0) - goto close_signalfd; + ret = event_loop_add(worker->event_loop, worker->signalfd, POLLIN, worker_set_stopping, worker); + if (ret < 0) + goto close_signalfd; - ret = libgit_init(); - if (ret < 0) - goto close_signalfd; + ret = libgit_init(); + if (ret < 0) + goto close_signalfd; - *_worker = worker; - return ret; + *_worker = worker; + return ret; close_signalfd: - signalfd_destroy(worker->signalfd); + signalfd_destroy(worker->signalfd); destroy_event_loop: - event_loop_destroy(worker->event_loop); + event_loop_destroy(worker->event_loop); destroy_cmd_dispatcher: - cmd_dispatcher_destroy(worker->cmd_dispatcher); + cmd_dispatcher_destroy(worker->cmd_dispatcher); free_settings: - worker_settings_destroy(worker->settings); + worker_settings_destroy(worker->settings); free: - free(worker); + free(worker); - return ret; + return ret; } void worker_destroy(struct worker* worker) { - log("Shutting down\n"); + log("Shutting down\n"); - libgit_shutdown(); - signalfd_destroy(worker->signalfd); - event_loop_destroy(worker->event_loop); - cmd_dispatcher_destroy(worker->cmd_dispatcher); - worker_settings_destroy(worker->settings); - free(worker); + libgit_shutdown(); + signalfd_destroy(worker->signalfd); + event_loop_destroy(worker->event_loop); + cmd_dispatcher_destroy(worker->cmd_dispatcher); + worker_settings_destroy(worker->settings); + free(worker); } static int worker_do_run(struct worker* worker) { - int ret = 0; + int ret = 0; - struct process_output* result = NULL; - ret = process_output_create(&result); - if (ret < 0) - return ret; + struct process_output* result = NULL; + ret = process_output_create(&result); + if (ret < 0) + return ret; - ret = ci_run_git_repo(run_get_repo_url(worker->run), run_get_repo_rev(worker->run), result); - if (ret < 0) { - log_err("Run failed with an error\n"); - goto free_output; - } + ret = ci_run_git_repo(run_get_repo_url(worker->run), run_get_repo_rev(worker->run), result); + if (ret < 0) { + log_err("Run failed with an error\n"); + goto free_output; + } - process_output_dump(result); + process_output_dump(result); - struct jsonrpc_request* finished_request = NULL; + struct jsonrpc_request* finished_request = NULL; - ret = request_create_finished_run(&finished_request, run_get_id(worker->run), result); - if (ret < 0) - goto free_output; + ret = request_create_finished_run(&finished_request, run_get_id(worker->run), result); + if (ret < 0) + goto free_output; - ret = net_connect(worker->settings->host, worker->settings->port); - if (ret < 0) - goto free_request; - int fd = ret; + ret = net_connect(worker->settings->host, worker->settings->port); + if (ret < 0) + goto free_request; + int fd = ret; - ret = jsonrpc_request_send(finished_request, fd); - if (ret < 0) - goto close_conn; + ret = jsonrpc_request_send(finished_request, fd); + if (ret < 0) + goto close_conn; close_conn: - net_close(fd); + net_close(fd); free_request: - jsonrpc_request_destroy(finished_request); + jsonrpc_request_destroy(finished_request); free_output: - process_output_destroy(result); + process_output_destroy(result); - run_destroy(worker->run); + run_destroy(worker->run); - return ret; + return ret; } static int worker_get_run(struct worker* worker) { - int ret = 0, fd = -1; + int ret = 0, fd = -1; - ret = net_connect(worker->settings->host, worker->settings->port); - if (ret < 0) - return ret; - fd = ret; + ret = net_connect(worker->settings->host, worker->settings->port); + if (ret < 0) + return ret; + fd = ret; - struct jsonrpc_request* new_worker_request = NULL; - ret = request_create_new_worker(&new_worker_request); - if (ret < 0) - goto close; + struct jsonrpc_request* new_worker_request = NULL; + ret = request_create_new_worker(&new_worker_request); + if (ret < 0) + goto close; - ret = jsonrpc_request_send(new_worker_request, fd); - jsonrpc_request_destroy(new_worker_request); - if (ret < 0) - goto close; + ret = jsonrpc_request_send(new_worker_request, fd); + jsonrpc_request_destroy(new_worker_request); + if (ret < 0) + goto close; - ret = event_loop_add_once( - worker->event_loop, fd, POLLIN, cmd_dispatcher_handle_event, worker->cmd_dispatcher); - if (ret < 0) - goto close; + ret = event_loop_add_once( + worker->event_loop, fd, POLLIN, cmd_dispatcher_handle_event, worker->cmd_dispatcher); + if (ret < 0) + goto close; - log("Waiting for a new command\n"); + log("Waiting for a new command\n"); - ret = event_loop_run(worker->event_loop); - if (ret < 0) - goto close; + ret = event_loop_run(worker->event_loop); + if (ret < 0) + goto close; close: - net_close(fd); + net_close(fd); - return ret; + return ret; } int worker_main(struct worker* worker) { - int ret = 0; + int ret = 0; - while (1) { - ret = worker_get_run(worker); - if (ret < 0) - return ret; + while (1) { + ret = worker_get_run(worker); + if (ret < 0) + return ret; - if (worker->stopping) - break; + if (worker->stopping) + break; - ret = worker_do_run(worker); - if (ret < 0) - return ret; - } + ret = worker_do_run(worker); + if (ret < 0) + return ret; + } - return ret; + return ret; } diff --git a/src/worker.h b/src/worker.h index b9450d6..653a058 100644 --- a/src/worker.h +++ b/src/worker.h @@ -9,8 +9,8 @@ #define __WORKER_H__ struct settings { - const char* host; - const char* port; + const char* host; + const char* port; }; struct worker; diff --git a/src/worker_main.c b/src/worker_main.c index da6db25..7d076c7 100644 --- a/src/worker_main.c +++ b/src/worker_main.c @@ -14,23 +14,23 @@ #include <unistd.h> static struct settings default_settings(void) { - struct settings settings = { - .host = default_host, - .port = default_port, - }; - return settings; + struct settings settings = { + .host = default_host, + .port = default_port, + }; + return settings; } const char* get_usage_string(void) { - return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT]"; + return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT]"; } static int parse_settings(struct settings* settings, int argc, char* argv[]) { - int opt, longind; + int opt, longind; - *settings = default_settings(); + *settings = default_settings(); - /* clang-format off */ + /* clang-format off */ static struct option long_options[] = { {"help", no_argument, 0, 'h'}, {"version", no_argument, 0, 'V'}, @@ -39,53 +39,53 @@ static int parse_settings(struct settings* settings, int argc, char* argv[]) { {"port", required_argument, 0, 'p'}, {0, 0, 0, 0}, }; - /* clang-format on */ + /* clang-format on */ - while ((opt = getopt_long(argc, argv, "hVvH:p:", long_options, &longind)) != -1) { - switch (opt) { - case 'h': - exit_with_usage(0); - break; - case 'V': - exit_with_version(); - break; - case 'v': - g_log_lvl = LOG_LVL_DEBUG; - break; - case 'H': - settings->host = optarg; - break; - case 'p': - settings->port = optarg; - break; - default: - exit_with_usage(1); - break; - } - } + while ((opt = getopt_long(argc, argv, "hVvH:p:", long_options, &longind)) != -1) { + switch (opt) { + case 'h': + exit_with_usage(0); + break; + case 'V': + exit_with_version(); + break; + case 'v': + g_log_lvl = LOG_LVL_DEBUG; + break; + case 'H': + settings->host = optarg; + break; + case 'p': + settings->port = optarg; + break; + default: + exit_with_usage(1); + break; + } + } - return 0; + return 0; } int main(int argc, char* argv[]) { - struct settings settings; - struct worker* worker = NULL; - int ret = 0; + struct settings settings; + struct worker* worker = NULL; + int ret = 0; - ret = parse_settings(&settings, argc, argv); - if (ret < 0) - return ret; + ret = parse_settings(&settings, argc, argv); + if (ret < 0) + return ret; - ret = worker_create(&worker, &settings); - if (ret < 0) - return ret; + ret = worker_create(&worker, &settings); + if (ret < 0) + return ret; - ret = worker_main(worker); - if (ret < 0) - goto destroy_worker; + ret = worker_main(worker); + if (ret < 0) + goto destroy_worker; destroy_worker: - worker_destroy(worker); + worker_destroy(worker); - return ret; + return ret; } diff --git a/src/worker_queue.c b/src/worker_queue.c index 84f9c50..13bdc1f 100644 --- a/src/worker_queue.c +++ b/src/worker_queue.c @@ -14,60 +14,60 @@ #include <sys/queue.h> struct worker { - int fd; - SIMPLEQ_ENTRY(worker) entries; + int fd; + SIMPLEQ_ENTRY(worker) entries; }; int worker_create(struct worker** _entry, int fd) { - struct worker* entry = malloc(sizeof(struct worker)); - if (!entry) { - log_errno("malloc"); - return -1; - } + struct worker* entry = malloc(sizeof(struct worker)); + if (!entry) { + log_errno("malloc"); + return -1; + } - entry->fd = fd; + entry->fd = fd; - *_entry = entry; - return 0; + *_entry = entry; + return 0; } void worker_destroy(struct worker* entry) { - net_close(entry->fd); - free(entry); + net_close(entry->fd); + free(entry); } int worker_get_fd(const struct worker* entry) { - return entry->fd; + return entry->fd; } void worker_queue_create(struct worker_queue* queue) { - SIMPLEQ_INIT(queue); + SIMPLEQ_INIT(queue); } void worker_queue_destroy(struct worker_queue* queue) { - struct worker* entry1 = SIMPLEQ_FIRST(queue); - while (entry1) { - struct worker* entry2 = SIMPLEQ_NEXT(entry1, entries); - worker_destroy(entry1); - entry1 = entry2; - } - SIMPLEQ_INIT(queue); + struct worker* entry1 = SIMPLEQ_FIRST(queue); + while (entry1) { + struct worker* entry2 = SIMPLEQ_NEXT(entry1, entries); + worker_destroy(entry1); + entry1 = entry2; + } + SIMPLEQ_INIT(queue); } int worker_queue_is_empty(const struct worker_queue* queue) { - return SIMPLEQ_EMPTY(queue); + return SIMPLEQ_EMPTY(queue); } void worker_queue_add_first(struct worker_queue* queue, struct worker* entry) { - SIMPLEQ_INSERT_HEAD(queue, entry, entries); + SIMPLEQ_INSERT_HEAD(queue, entry, entries); } void worker_queue_add_last(struct worker_queue* queue, struct worker* entry) { - SIMPLEQ_INSERT_TAIL(queue, entry, entries); + SIMPLEQ_INSERT_TAIL(queue, entry, entries); } struct worker* worker_queue_remove_first(struct worker_queue* queue) { - struct worker* entry = SIMPLEQ_FIRST(queue); - SIMPLEQ_REMOVE_HEAD(queue, entries); - return entry; + struct worker* entry = SIMPLEQ_FIRST(queue); + SIMPLEQ_REMOVE_HEAD(queue, entries); + return entry; } |