#include <sys/types.h>
#include <sys/ioctl.h>
#include <sys/poll.h>
+#include <sys/time.h>
#include <signal.h>
#include <errno.h>
#include <ctype.h>
+#include <fcntl.h>
#include <vector>
#include <string>
#include <map>
pfd.events = POLLIN;
int nfds = poll(&pfd, 1, 50);
- if (nfds == 0 || (nfds == -1 && errno == EAGAIN)) {
+ if (nfds == 0 || (nfds == -1 && errno == EINTR)) {
continue;
}
if (nfds == -1) {
return NULL;
}
+struct StatsThreadParameters {
+ string stats_file;
+ int stats_interval;
+};
+
+void *stats_thread_run(void *arg)
+{
+ const StatsThreadParameters *parms = reinterpret_cast<StatsThreadParameters *>(arg);
+ while (!hupped) {
+ int fd;
+ FILE *fp;
+ time_t now;
+ vector<ClientStats> client_stats;
+
+ // Open a new, temporary file.
+ char *filename = strdup((parms->stats_file + ".new.XXXXXX").c_str());
+ fd = mkostemp(filename, O_WRONLY);
+ if (fd == -1) {
+ perror(filename);
+ free(filename);
+ goto sleep;
+ }
+
+ fp = fdopen(fd, "w");
+ if (fp == NULL) {
+ perror("fdopen");
+ close(fd);
+ unlink(filename);
+ free(filename);
+ goto sleep;
+ }
+
+ now = time(NULL);
+ client_stats = servers->get_client_stats();
+ for (size_t i = 0; i < client_stats.size(); ++i) {
+ fprintf(fp, "%s %s %d %llu\n",
+ client_stats[i].remote_addr.c_str(),
+ client_stats[i].stream_id.c_str(),
+ int(now - client_stats[i].connect_time),
+ (long long unsigned)(client_stats[i].bytes_sent));
+ }
+ if (fclose(fp) == EOF) {
+ perror("fclose");
+ unlink(filename);
+ free(filename);
+ goto sleep;
+ }
+
+ if (rename(filename, parms->stats_file.c_str()) == -1) {
+ perror("rename");
+ unlink(filename);
+ }
+
+sleep:
+ int left_to_sleep = parms->stats_interval;
+ do {
+ left_to_sleep = sleep(left_to_sleep);
+ } while (left_to_sleep > 0 && !hupped);
+ }
+ return NULL;
+}
+
// Serialize the given state to a file descriptor, and return the (still open)
// descriptor.
int make_tempfile(const CubemapStateProto &state)
{
fprintf(stderr, "\nCubemap starting.\n");
+ struct timeval serialize_start;
+ bool is_reexec = false;
+
string config_filename = (argc == 1) ? "cubemap.config" : argv[1];
vector<ConfigLine> config = parse_config(config_filename);
- int port = fetch_config_int(config, "port", 1, 65535);
- int num_servers = fetch_config_int(config, "num_servers", 1, 20000); // Insanely high max limit.
+ int port = fetch_config_int(config, "port", 1, 65535, PARAMATER_MANDATORY);
+ int num_servers = fetch_config_int(config, "num_servers", 1, 20000, PARAMATER_MANDATORY); // Insanely high max limit.
servers = new ServerPool(num_servers);
set<string> deserialized_stream_ids;
map<string, Input *> deserialized_inputs;
if (argc == 4 && strcmp(argv[2], "-state") == 0) {
+ is_reexec = true;
+
fprintf(stderr, "Deserializing state from previous process... ");
int state_fd = atoi(argv[3]);
CubemapStateProto loaded_state = read_tempfile(state_fd);
+ serialize_start.tv_sec = loaded_state.serialize_start_sec();
+ serialize_start.tv_usec = loaded_state.serialize_start_usec();
+
// Deserialize the streams.
for (int i = 0; i < loaded_state.streams_size(); ++i) {
servers->add_stream_from_serialized(loaded_state.streams(i));
server_sock = create_server_socket(port);
}
+ // See if the user wants stats.
+ string stats_file = fetch_config_string(config, "stats_file", PARAMETER_OPTIONAL);
+ int stats_interval = fetch_config_int(config, "stats_interval", 1, INT_MAX, PARAMETER_OPTIONAL, -1);
+ if (stats_interval != -1 && stats_file.empty()) {
+ fprintf(stderr, "WARNING: 'stats_interval' given, but no 'stats_file'. No statistics will be written.\n");
+ }
+
servers->run();
pthread_t acceptor_thread;
// All deserialized inputs should now have been taken care of, one way or the other.
assert(deserialized_inputs.empty());
+ // Start writing statistics.
+ pthread_t stats_thread;
+ StatsThreadParameters stats_parameters; // Must live for as long as the stats thread does.
+ if (!stats_file.empty()) {
+ stats_parameters.stats_file = stats_file;
+ stats_parameters.stats_interval = stats_interval;
+ pthread_create(&stats_thread, NULL, stats_thread_run, &stats_parameters);
+ }
+
signal(SIGHUP, hup);
+
+ struct timeval server_start;
+ gettimeofday(&server_start, NULL);
+ if (is_reexec) {
+ // Measure time from we started deserializing (below) to now, when basically everything
+ // is up and running. This is, in other words, a conservative estimate of how long our
+ // “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);
+ }
while (!hupped) {
usleep(100000);
}
// OK, we've been HUPed. Time to shut down everything, serialize, and re-exec.
+ gettimeofday(&serialize_start, NULL);
+
+ if (!stats_file.empty()) {
+ pthread_kill(stats_thread, SIGHUP);
+ if (pthread_join(stats_thread, NULL) == -1) {
+ perror("pthread_join");
+ exit(1);
+ }
+ }
+ pthread_kill(acceptor_thread, SIGHUP);
if (pthread_join(acceptor_thread, NULL) == -1) {
perror("pthread_join");
exit(1);
}
CubemapStateProto state;
+ state.set_serialize_start_sec(serialize_start.tv_sec);
+ state.set_serialize_start_usec(serialize_start.tv_usec);
state.set_server_sock(server_sock);
state.set_port(port);
projects / cubemap / blobdiff