#include <stdint.h>
#include <assert.h>
#include <arpa/inet.h>
-#include <curl/curl.h>
#include <sys/socket.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/ioctl.h>
-#include <sys/epoll.h>
+#include <sys/poll.h>
#include <signal.h>
#include <errno.h>
#include <ctype.h>
#include <vector>
#include <string>
#include <map>
+#include <set>
#include "metacube.h"
#include "parse.h"
exit(1);
}
+ // Set as non-blocking, so the acceptor thread can notice that we want to shut it down.
+ if (ioctl(server_sock, FIONBIO, &one) == -1) {
+ perror("ioctl(FIONBIO)");
+ exit(1);
+ }
+
sockaddr_in6 addr;
memset(&addr, 0, sizeof(addr));
addr.sin6_family = AF_INET6;
void *acceptor_thread_run(void *arg)
{
int server_sock = int(intptr_t(arg));
- int num_accepted = 0;
- for ( ;; ) {
+ while (!hupped) {
+ // Since we are non-blocking, we need to wait for the right state first.
+ // Wait up to 50 ms, then check hupped.
+ pollfd pfd;
+ pfd.fd = server_sock;
+ pfd.events = POLLIN;
+
+ int nfds = poll(&pfd, 1, 50);
+ if (nfds == 0 || (nfds == -1 && errno == EAGAIN)) {
+ continue;
+ }
+ if (nfds == -1) {
+ perror("poll");
+ usleep(100000);
+ continue;
+ }
+
sockaddr_in6 addr;
socklen_t addrlen = sizeof(addr);
}
if (sock == -1) {
perror("accept");
- exit(1);
+ usleep(100000);
+ continue;
}
// Set the socket as nonblocking.
// Pick a server, round-robin, and hand over the socket to it.
servers->add_client(sock);
- ++num_accepted;
}
+ return NULL;
}
// Serialize the given state to a file descriptor, and return the (still open)
servers = new ServerPool(num_servers);
int server_sock = -1, old_port = -1;
+ set<string> deserialized_stream_ids;
if (argc == 4 && strcmp(argv[2], "-state") == 0) {
fprintf(stderr, "Deserializing state from previous process... ");
int state_fd = atoi(argv[3]);
CubemapStateProto loaded_state = read_tempfile(state_fd);
// Deserialize the streams.
- // TODO: Pick up new streams from the configuration file.
for (int i = 0; i < loaded_state.streams_size(); ++i) {
servers->add_stream_from_serialized(loaded_state.streams(i));
+ deserialized_stream_ids.insert(loaded_state.streams(i).stream_id());
}
// Put back the existing clients. It doesn't matter which server we
old_port = loaded_state.port();
fprintf(stderr, "done.\n");
- } else {
- // Find all streams in the configuration file, and create them.
- for (unsigned i = 0; i < config.size(); ++i) {
- if (config[i].keyword != "stream") {
- continue;
- }
- if (config[i].arguments.size() != 1) {
- fprintf(stderr, "ERROR: 'stream' takes exactly one argument\n");
- exit(1);
- }
- string stream_id = config[i].arguments[0];
+ }
+
+ // Find all streams in the configuration file, and create them.
+ set<string> expecting_stream_ids = deserialized_stream_ids;
+ for (unsigned i = 0; i < config.size(); ++i) {
+ if (config[i].keyword != "stream") {
+ continue;
+ }
+ if (config[i].arguments.size() != 1) {
+ fprintf(stderr, "ERROR: 'stream' takes exactly one argument\n");
+ exit(1);
+ }
+ string stream_id = config[i].arguments[0];
+ if (deserialized_stream_ids.count(stream_id) == 0) {
servers->add_stream(stream_id);
}
+ expecting_stream_ids.erase(stream_id);
+ }
+
+ // Warn about any servers we've lost.
+ // TODO: Make an option (delete=yes?) to actually shut down streams.
+ for (set<string>::const_iterator stream_it = expecting_stream_ids.begin();
+ stream_it != expecting_stream_ids.end();
+ ++stream_it) {
+ fprintf(stderr, "WARNING: stream '%s' disappeared from the configuration file.\n",
+ stream_it->c_str());
+ fprintf(stderr, " It will not be deleted, but clients will not get any new inputs.\n");
}
// Open a new server socket if we do not already have one, or if we changed ports.
// OK, we've been HUPed. Time to shut down everything, serialize, and re-exec.
for (size_t i = 0; i < inputs.size(); ++i) {
inputs[i]->stop();
- delete inputs[i]; // TODO: serialize instead of using libcurl.
+ delete inputs[i]; // TODO: Serialize.
+ }
+
+ if (pthread_join(acceptor_thread, NULL) == -1) {
+ perror("pthread_join");
+ exit(1);
}
CubemapStateProto state;