diff options
Diffstat (limited to 'src/server.c')
| -rw-r--r-- | src/server.c | 537 |
1 files changed, 268 insertions, 269 deletions
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); } |