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

diff --git a/cubemap.cpp b/cubemap.cpp
index 692d48baf1d9e507d36926aa10a7f35f7d8847b1..5ef003640ff1cc5780c6cd79c43687324193fcc6 100644 (file)
--- a/cubemap.cpp
+++ b/cubemap.cpp
@@ -9,6 +9,7 @@
 #include <sys/types.h>
 #include <sys/ioctl.h>
 #include <sys/epoll.h>
 #include <sys/types.h>
 #include <sys/ioctl.h>
 #include <sys/epoll.h>

+#include <signal.h>

 #include <errno.h>
 #include <vector>
 #include <string>
 #include <errno.h>
 #include <vector>
 #include <string>


@@ -17,6 +18,7 @@
 #include "metacube.h"
 #include "server.h"
 #include "input.h"
 #include "metacube.h"
 #include "server.h"
 #include "input.h"

+#include "state.pb.h"

 
 #define NUM_SERVERS 4
 #define STREAM_ID "stream"
 
 #define NUM_SERVERS 4
 #define STREAM_ID "stream"


@@ -26,6 +28,12 @@
 using namespace std;
 
 Server *servers = NULL;
 using namespace std;
 
 Server *servers = NULL;

+volatile bool hupped = false;
+
+void hup(int ignored)
+{
+       hupped = true;
+}

 
 int create_server_socket(int port)
 {
 
 int create_server_socket(int port)
 {


@@ -110,11 +118,32 @@ int main(int argc, char **argv)
        pthread_t acceptor_thread;
        pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast<void *>(server_sock));
 
        pthread_t acceptor_thread;
        pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast<void *>(server_sock));
 

-       Input input(STREAM_ID);
-       input.run(STREAM_URL);
+       Input input(STREAM_ID, STREAM_URL);
+       input.run();
+
+       signal(SIGHUP, hup);
+
+       while (!hupped) {
+               usleep(100000);
+       }
+
+       input.stop();

 
 

+       CubemapStateProto state;

        for (int i = 0; i < NUM_SERVERS; ++i) {
                servers[i].stop();
        for (int i = 0; i < NUM_SERVERS; ++i) {
                servers[i].stop();

+
+               CubemapStateProto local_state = servers[i].serialize();
+
+               // The stream state should be identical between the servers, so we only store it once.
+               if (i == 0) {
+                       state.mutable_streams()->MergeFrom(local_state.streams());
+               }
+               for (int j = 0; j < local_state.clients_size(); ++j) {
+                       state.add_clients()->MergeFrom(local_state.clients(j));
+               }

        }
        delete[] servers;
        }
        delete[] servers;

+
+       printf("SERIALIZED: [%s]\n", state.DebugString().c_str());

 }
 }

diff --git a/input.cpp b/input.cpp
index d36919407077c60aff7f34eabb5a74dc57f15aab..f9f879b6c2a393f6671b42fb25e4462854e04711 100644 (file)
--- a/input.cpp
+++ b/input.cpp
@@ -15,6 +15,7 @@
 #include <map>
 
 #include "metacube.h"
 #include <map>
 
 #include "metacube.h"

+#include "mutexlock.h"

 #include "input.h"
 #include "server.h"
 
 #include "input.h"
 #include "server.h"
 


@@ -22,13 +23,42 @@ using namespace std;
 
 extern Server *servers;
 
 
 extern Server *servers;
 

-Input::Input(const string &stream_id)
+Input::Input(const string &stream_id, const string &url)

        : stream_id(stream_id),
        : stream_id(stream_id),

+         url(url),

          has_metacube_header(false)
 {
 }
 
          has_metacube_header(false)
 {
 }
 

-void Input::run(const string &url)
+void Input::run()
+{
+       should_stop = false;
+       
+       // Joinable is already the default, but it's good to be certain.
+       pthread_attr_t attr;
+       pthread_attr_init(&attr);
+       pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+       pthread_create(&worker_thread, &attr, Input::do_work_thunk, this);
+}
+
+void Input::stop()
+{
+       should_stop = true;
+
+       if (pthread_join(worker_thread, NULL) == -1) {
+               perror("pthread_join");
+               exit(1);
+       }
+}
+
+void *Input::do_work_thunk(void *arg)
+{
+       Input *input = static_cast<Input *>(arg);
+       input->do_work();
+       return NULL;
+}
+
+void Input::do_work()

 {
        CURL *curl = curl_easy_init();
        curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
 {
        CURL *curl = curl_easy_init();
        curl_easy_setopt(curl, CURLOPT_URL, url.c_str());


@@ -40,6 +70,10 @@ void Input::run(const string &url)
 size_t Input::curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata)
 {
        Input *input = static_cast<Input *>(userdata);
 size_t Input::curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata)
 {
        Input *input = static_cast<Input *>(userdata);

+       if (input->should_stop) {
+               return 0;
+       }
+

        size_t bytes = size * nmemb;
        input->curl_callback(ptr, bytes);       
        return bytes;
        size_t bytes = size * nmemb;
        input->curl_callback(ptr, bytes);       
        return bytes;

diff --git a/input.h b/input.h
index c293071ecfa7a9f7aae9562905841cac690dc040..508eb59ae68cac1ba029c88caae5f67bbff73000 100644 (file)
--- a/input.h
+++ b/input.h
@@ -6,13 +6,21 @@
 
 class Input {
 public:
 
 class Input {
 public:

-       Input(const std::string &stream_id);
+       Input(const std::string &stream_id, const std::string &url);

 
        // Connect to the given URL and start streaming.
 
        // Connect to the given URL and start streaming.

-       // WARNING: Currently this blocks; it does not run in a separate thread!
-       void run(const std::string &url);
+       void run();
+
+       // Stop streaming. NOTE: Does not currently work!
+       void stop();

 
 private:
 
 private:

+       // Recovers the this pointer and calls do_work().
+       static void *do_work_thunk(void *arg);
+
+       // Actually does the download.
+       void do_work();
+

        // Recovers the this pointer and calls curl_callback().
        static size_t curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata);
 
        // Recovers the this pointer and calls curl_callback().
        static size_t curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata);
 


@@ -26,6 +34,7 @@ private:
        void drop_pending_data(size_t num_bytes);
 
        std::string stream_id;
        void drop_pending_data(size_t num_bytes);
 
        std::string stream_id;

+       std::string url;

 
        // Data we have received but not fully processed yet.
        std::vector<char> pending_data;
 
        // Data we have received but not fully processed yet.
        std::vector<char> pending_data;


@@ -33,6 +42,11 @@ private:
        // If <pending_data> starts with a Metacube header,
        // this is true.
        bool has_metacube_header;
        // If <pending_data> starts with a Metacube header,
        // this is true.
        bool has_metacube_header;

+
+       pthread_t worker_thread;
+
+       // Whether we should stop or not.
+       volatile bool should_stop;

 };
 
 #endif  // !defined(_INPUT_H)
 };
 
 #endif  // !defined(_INPUT_H)

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;
 }
 
        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);
 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);
 }
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
        pthread_create(&worker_thread, &attr, Server::do_work_thunk, this);
 }

-
+       

 void Server::stop()
 {
        {
 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);
 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;
        // 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)");
        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);
 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);
 }
        
 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)
 }
        
 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);
        }
 
        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;
 
        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;
                // 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.
                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));
                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;
                }
 
                        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" +
 
        // 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;
 
        // 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) {
        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;
 {
        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) {
        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;
        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)");
                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();
 }
        }
        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;
+}

diff --git a/server.h b/server.h
index d0e3d6a75a705b0ed9a2a16624155b0227800828..99009600a46f0401a325ced52e04f542c57b602f 100644 (file)
--- a/server.h
+++ b/server.h
@@ -13,6 +13,8 @@
 #define MAX_CLIENT_REQUEST 16384
 
 class ClientProto;
 #define MAX_CLIENT_REQUEST 16384
 
 class ClientProto;

+class CubemapStateProto;
+class StreamProto;

 
 struct Client {
        Client() {}
 
 struct Client {
        Client() {}


@@ -49,15 +51,27 @@ struct Client {
 };
 
 struct Stream {
 };
 
 struct Stream {

+       Stream(const std::string &stream_id);
+       ~Stream();
+
+       // Serialization/deserialization.
+       Stream(const StreamProto &serialized);
+       StreamProto serialize() const;
+
+       std::string stream_id;
+

        // The HTTP response header, plus the video stream header (if any).
        std::string header;
 
        // The stream data itself, stored in a circular buffer.
        // The HTTP response header, plus the video stream header (if any).
        std::string header;
 
        // The stream data itself, stored in a circular buffer.

-       char data[BACKLOG_SIZE];
+       char *data;

 
        // How many bytes <data> contains. Can very well be larger than BACKLOG_SIZE,
        // since the buffer wraps.
        size_t data_size;
 
        // How many bytes <data> contains. Can very well be larger than BACKLOG_SIZE,
        // since the buffer wraps.
        size_t data_size;

+
+private:
+       Stream(const Stream& other);

 };
 
 class Server {
 };
 
 class Server {


@@ -70,6 +84,8 @@ public:
        // Stop the thread.
        void stop();
 
        // Stop the thread.
        void stop();
 

+       CubemapStateProto serialize() const;
+

        void add_client(int sock);
        void add_stream(const std::string &stream_id);
        void set_header(const std::string &stream_id, const std::string &header);
        void add_client(int sock);
        void add_stream(const std::string &stream_id);
        void set_header(const std::string &stream_id, const std::string &header);


@@ -85,7 +101,7 @@ private:
        bool should_stop;       
 
        // Map from stream ID to stream.
        bool should_stop;       
 
        // Map from stream ID to stream.

-       std::map<std::string, Stream> streams;
+       std::map<std::string, Stream *> streams;

 
        // Map from file descriptor to client.
        std::map<int, Client> clients;
 
        // Map from file descriptor to client.
        std::map<int, Client> clients;


@@ -121,6 +137,9 @@ private:
 
        // We have more data, so mark all clients that are sleeping as ready to go.
        void wake_up_all_clients();
 
        // We have more data, so mark all clients that are sleeping as ready to go.
        void wake_up_all_clients();

+
+       // TODO: This function should probably die.
+       Stream *find_stream(const std::string &stream_id);

 };
 
 #endif  // !defined(_SERVER_H)
 };
 
 #endif  // !defined(_SERVER_H)