Serialize and die on SIGHUP (no deserialization yet). Required a whole bunch of Valgr...

[cubemap] / server.cpp

diff --git a/server.cpp b/server.cpp
index 69615e705aba182b9fbbf168c56ea5c78931695c..c80f2da2e2c354729702ef04ba9f7068ed5ec14c 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -55,6 +55,37 @@ ClientProto Client::serialize() const
        return serialized;
 }
 
+Stream::Stream(const string &stream_id)
+       : stream_id(stream_id),
+         data(new char[BACKLOG_SIZE]),
+         data_size(0)
+{
+       memset(data, 0, BACKLOG_SIZE);
+}
+
+Stream::~Stream()
+{
+       delete[] data;
+}
+
+Stream::Stream(const StreamProto &serialized)
+       : header(serialized.header()),
+         data(new char[BACKLOG_SIZE]),
+         data_size(serialized.data_size())
+{
+       assert(serialized.data().size() == BACKLOG_SIZE);
+       memcpy(data, serialized.data().data(), BACKLOG_SIZE);
+}
+
+StreamProto Stream::serialize() const
+{
+       StreamProto serialized;
+       serialized.set_header(header);
+       serialized.set_data(string(data, data + BACKLOG_SIZE));
+       serialized.set_data_size(data_size);
+       return serialized;
+}
+
 Server::Server()
 {
        pthread_mutex_init(&mutex, NULL);
@@ -76,7 +107,7 @@ void Server::run()
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
        pthread_create(&worker_thread, &attr, Server::do_work_thunk, this);
 }
-
+       
 void Server::stop()
 {
        {
@@ -126,7 +157,23 @@ void Server::do_work()
                }
        }
 }
-       
+
+CubemapStateProto Server::serialize() const
+{
+       CubemapStateProto serialized;
+       for (map<int, Client>::const_iterator client_it = clients.begin();
+            client_it != clients.end();
+            ++client_it) {
+               serialized.add_clients()->MergeFrom(client_it->second.serialize());
+       }
+       for (map<string, Stream *>::const_iterator stream_it = streams.begin();
+            stream_it != streams.end();
+            ++stream_it) {
+               serialized.add_streams()->MergeFrom(stream_it->second->serialize());
+       }
+       return serialized;
+}
+
 void Server::add_client(int sock)
 {
        MutexLock lock(&mutex);
@@ -135,6 +182,7 @@ void Server::add_client(int sock)
        // Start listening on data from this socket.
        epoll_event ev;
        ev.events = EPOLLIN | EPOLLRDHUP;
+       ev.data.u64 = 0;  // Keep Valgrind happy.
        ev.data.fd = sock;
        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
                perror("epoll_ctl(EPOLL_CTL_ADD)");
@@ -145,14 +193,13 @@ void Server::add_client(int sock)
 void Server::add_stream(const string &stream_id)
 {
        MutexLock lock(&mutex);
-       streams.insert(make_pair(stream_id, Stream()));
+       streams.insert(make_pair(stream_id, new Stream(stream_id)));
 }
        
 void Server::set_header(const string &stream_id, const string &header)
 {
        MutexLock lock(&mutex);
-       assert(streams.count(stream_id) != 0);
-       streams[stream_id].header = header;
+       find_stream(stream_id)->header = header;
 }
        
 void Server::add_data(const string &stream_id, const char *data, size_t bytes)
@@ -162,8 +209,7 @@ void Server::add_data(const string &stream_id, const char *data, size_t bytes)
        }
 
        MutexLock lock(&mutex);
-       assert(streams.count(stream_id) != 0);
-       Stream *stream = &streams[stream_id];
+       Stream *stream = find_stream(stream_id);
        size_t pos = stream->data_size % BACKLOG_SIZE;
        stream->data_size += bytes;
 
@@ -253,19 +299,19 @@ void Server::process_client(Client *client)
                // Start sending from the end. In other words, we won't send any of the backlog,
                // but we'll start sending immediately as we get data.
                client->state = Client::SENDING_DATA;
-               client->bytes_sent = streams[client->stream_id].data_size;
+               client->bytes_sent = find_stream(client->stream_id)->data_size;
                break;
        }
        case Client::SENDING_DATA: {
                // See if there's some data we've lost. Ideally, we should drop to a block boundary,
                // but resync will be the mux's problem.
-               const Stream &stream = streams[client->stream_id];
+               const Stream &stream = *find_stream(client->stream_id);
                size_t bytes_to_send = stream.data_size - client->bytes_sent;
                if (bytes_to_send > BACKLOG_SIZE) {
                        fprintf(stderr, "WARNING: fd %d lost %lld bytes, maybe too slow connection\n",
                                client->sock,
                                (long long int)(bytes_to_send - BACKLOG_SIZE));
-                       client->bytes_sent = streams[client->stream_id].data_size - BACKLOG_SIZE;
+                       client->bytes_sent = find_stream(client->stream_id)->data_size - BACKLOG_SIZE;
                        bytes_to_send = BACKLOG_SIZE;
                }
 
@@ -313,13 +359,14 @@ void Server::parse_request(Client *client)
 
        // Construct the header.
        client->header = "HTTP/1.0 200 OK\r\n  Content-type: video/x-flv\r\nCache-Control: no-cache\r\nContent-type: todo/fixme\r\n\r\n" +
-               streams[client->stream_id].header;
+               find_stream(client->stream_id)->header;
 
        // Switch states.
        client->state = Client::SENDING_HEADER;
 
        epoll_event ev;
        ev.events = EPOLLOUT | EPOLLRDHUP;
+       ev.data.u64 = 0;  // Keep Valgrind happy.
        ev.data.fd = client->sock;
 
        if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
@@ -349,6 +396,7 @@ void Server::put_client_to_sleep(Client *client)
 {
        epoll_event ev;
        ev.events = EPOLLRDHUP;
+       ev.data.u64 = 0;  // Keep Valgrind happy.
        ev.data.fd = client->sock;
 
        if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
@@ -364,6 +412,7 @@ void Server::wake_up_all_clients()
        for (unsigned i = 0; i < sleeping_clients.size(); ++i) {
                epoll_event ev;
                ev.events = EPOLLOUT | EPOLLRDHUP;
+               ev.data.u64 = 0;  // Keep Valgrind happy.
                ev.data.fd = sleeping_clients[i];
                if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, sleeping_clients[i], &ev) == -1) {
                        perror("epoll_ctl(EPOLL_CTL_MOD)");
@@ -372,3 +421,10 @@ void Server::wake_up_all_clients()
        }
        sleeping_clients.clear();
 }
+       
+Stream *Server::find_stream(const string &stream_id)
+{
+       map<string, Stream *>::iterator it = streams.find(stream_id);
+       assert(it != streams.end());
+       return it->second;
+}