X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=main.cpp;h=4a5904b0715f1905817210138e3c9b7091329fae;hp=6c8051a351c471fac56d1e46cc695387ed8c382b;hb=0d72f384a1de672824298262ba5c427ec0aee2d6;hpb=ad06eb652a2cab501e5f676df08ac365b225795c diff --git a/main.cpp b/main.cpp index 6c8051a..4a5904b 100644 --- a/main.cpp +++ b/main.cpp @@ -1,39 +1,30 @@ -#include -#include -#include -#include #include +#include #include -#include -#include -#include -#include -#include -#include +#include +#include +#include +#include +#include #include -#include #include -#include -#include -#include -#include -#include -#include +#include #include #include +#include +#include +#include #include "acceptor.h" #include "config.h" +#include "input.h" +#include "log.h" #include "markpool.h" -#include "metacube.h" -#include "parse.h" -#include "server.h" #include "serverpool.h" -#include "input.h" +#include "state.pb.h" #include "stats.h" #include "util.h" #include "version.h" -#include "state.pb.h" using namespace std; @@ -118,7 +109,7 @@ vector create_inputs(const Config &config, if (deserialized_input_it != deserialized_inputs->end()) { input = deserialized_input_it->second; if (input->get_url() != src) { - fprintf(stderr, "INFO: Stream '%s' has changed URL from '%s' to '%s', restarting input.\n", + log(INFO, "Stream '%s' has changed URL from '%s' to '%s', restarting input.", stream_id.c_str(), input->get_url().c_str(), src.c_str()); input->close_socket(); delete input; @@ -129,7 +120,7 @@ vector create_inputs(const Config &config, if (input == NULL) { input = create_input(stream_id, src); if (input == NULL) { - fprintf(stderr, "ERROR: did not understand URL '%s', clients will not get any data.\n", + log(ERROR, "did not understand URL '%s', clients will not get any data.", src.c_str()); continue; } @@ -154,7 +145,9 @@ void create_streams(const Config &config, for (unsigned i = 0; i < config.streams.size(); ++i) { const StreamConfig &stream_config = config.streams[i]; if (deserialized_stream_ids.count(stream_config.stream_id) == 0) { - servers->add_stream(stream_config.stream_id); + servers->add_stream(stream_config.stream_id, stream_config.backlog_size); + } else { + servers->set_backlog_size(stream_config.stream_id, stream_config.backlog_size); } expecting_stream_ids.erase(stream_config.stream_id); @@ -170,9 +163,9 @@ void create_streams(const Config &config, stream_it != expecting_stream_ids.end(); ++stream_it) { string stream_id = *stream_it; - fprintf(stderr, "WARNING: stream '%s' disappeared from the configuration file.\n", - stream_id.c_str()); - fprintf(stderr, " It will not be deleted, but clients will not get any new inputs.\n"); + log(WARNING, "stream '%s' disappeared from the configuration file. " + "It will not be deleted, but clients will not get any new inputs.", + stream_id.c_str()); if (deserialized_inputs->count(stream_id) != 0) { delete (*deserialized_inputs)[stream_id]; deserialized_inputs->erase(stream_id); @@ -180,6 +173,22 @@ void create_streams(const Config &config, } } +void open_logs(const vector &log_destinations) +{ + for (size_t i = 0; i < log_destinations.size(); ++i) { + if (log_destinations[i].type == LogConfig::LOG_TYPE_FILE) { + add_log_destination_file(log_destinations[i].filename); + } else if (log_destinations[i].type == LogConfig::LOG_TYPE_CONSOLE) { + add_log_destination_console(); + } else if (log_destinations[i].type == LogConfig::LOG_TYPE_SYSLOG) { + add_log_destination_syslog(); + } else { + assert(false); + } + } + start_logging(); +} + bool dry_run_config(const std::string &argv0, const std::string &config_filename) { char *argv0_copy = strdup(argv0.c_str()); @@ -188,14 +197,14 @@ bool dry_run_config(const std::string &argv0, const std::string &config_filename pid_t pid = fork(); switch (pid) { case -1: - perror("fork()"); + log_perror("fork()"); free(argv0_copy); free(config_filename_copy); return false; case 0: // Child. execlp(argv0_copy, argv0_copy, "--test-config", config_filename_copy, NULL); - perror(argv0_copy); + log_perror(argv0_copy); _exit(1); default: // Parent. @@ -212,7 +221,7 @@ bool dry_run_config(const std::string &argv0, const std::string &config_filename } while (err == -1 && errno == EINTR); if (err == -1) { - perror("waitpid()"); + log_perror("waitpid()"); return false; } @@ -221,6 +230,9 @@ bool dry_run_config(const std::string &argv0, const std::string &config_filename int main(int argc, char **argv) { + signal(SIGHUP, hup); + signal(SIGPIPE, SIG_IGN); + // Parse options. int state_fd = -1; bool test_config = false; @@ -261,7 +273,10 @@ int main(int argc, char **argv) } start: - fprintf(stderr, "\nCubemap " SERVER_VERSION " starting.\n"); + // Open logs as soon as possible. + open_logs(config.log_destinations); + + log(NO_LEVEL, "Cubemap " SERVER_VERSION " starting."); servers = new ServerPool(config.num_servers); CubemapStateProto loaded_state; @@ -270,13 +285,13 @@ start: map deserialized_inputs; map deserialized_acceptors; if (state_fd != -1) { - fprintf(stderr, "Deserializing state from previous process... "); + log(INFO, "Deserializing state from previous process..."); string serialized; if (!read_tempfile(state_fd, &serialized)) { exit(1); } if (!loaded_state.ParseFromString(serialized)) { - fprintf(stderr, "ERROR: Failed deserialization of state.\n"); + log(ERROR, "Failed deserialization of state."); exit(1); } @@ -303,14 +318,12 @@ start: new Acceptor(loaded_state.acceptors(i)))); } - fprintf(stderr, "done.\n"); + log(INFO, "Deserialization done."); } // Find all streams in the configuration file, and create them. create_streams(config, deserialized_stream_ids, &deserialized_inputs); - servers->run(); - vector acceptors = create_acceptors(config, &deserialized_acceptors); vector inputs = create_inputs(config, &deserialized_inputs); @@ -323,6 +336,8 @@ start: for (int i = 0; i < loaded_state.clients_size(); ++i) { servers->add_client_from_serialized(loaded_state.clients(i)); } + + servers->run(); // Start writing statistics. StatsThread *stats_thread = NULL; @@ -331,8 +346,6 @@ start: stats_thread->run(); } - signal(SIGHUP, hup); - struct timeval server_start; gettimeofday(&server_start, NULL); if (state_fd != -1) { @@ -341,7 +354,7 @@ start: // “glitch” period was, not counting of course reconnects if the configuration changed. double glitch_time = server_start.tv_sec - serialize_start.tv_sec + 1e-6 * (server_start.tv_usec - serialize_start.tv_usec); - fprintf(stderr, "Re-exec happened in approx. %.0f ms.\n", glitch_time * 1000.0); + log(INFO, "Re-exec happened in approx. %.0f ms.", glitch_time * 1000.0); } while (!hupped) { @@ -362,7 +375,7 @@ start: } servers->stop(); - fprintf(stderr, "Serializing state and re-execing...\n"); + log(INFO, "Serializing state and re-execing..."); CubemapStateProto state = collect_state( serialize_start, acceptors, inputs, servers); string serialized; @@ -372,20 +385,26 @@ start: exit(1); } delete servers; + shut_down_logging(); if (!dry_run_config(argv[0], config_filename)) { - fprintf(stderr, "ERROR: %s --test-config failed. Restarting old version instead of new.\n", argv[0]); + open_logs(config.log_destinations); + log(ERROR, "%s --test-config failed. Restarting old version instead of new.", argv[0]); hupped = false; + shut_down_logging(); goto start; } + char buf[16]; sprintf(buf, "%d", state_fd); for ( ;; ) { execlp(argv[0], argv[0], config_filename.c_str(), "--state", buf, NULL); - perror("execlp"); - fprintf(stderr, "PANIC: re-exec of %s failed. Waiting 0.2 seconds and trying again...\n", argv[0]); + open_logs(config.log_destinations); + log_perror("execlp"); + log(ERROR, "re-exec of %s failed. Waiting 0.2 seconds and trying again...", argv[0]); + shut_down_logging(); usleep(200000); } }