git.sesse.net Git - cubemap/blob - server.h

   1 #ifndef _SERVER_H
   2 #define _SERVER_H 1
   3
   4 #include <stdint.h>
   5 #include <pthread.h>
   6 #include <sys/epoll.h>
   7 #include <string>
   8 #include <map>
   9 #include <vector>
  10
  11 #define BACKLOG_SIZE 1048576
  12 #define EPOLL_MAX_EVENTS 8192
  13 #define EPOLL_TIMEOUT_MS 20
  14 #define MAX_CLIENT_REQUEST 16384
  15
  16 class ClientProto;
  17 class CubemapStateProto;
  18 class StreamProto;
  19
  20 struct Client {
  21         Client() {}
  22         Client(int sock);
  23
  24         // Serialization/deserialization.
  25         Client(const ClientProto &serialized);
  26         ClientProto serialize() const;
  27
  28         // The file descriptor associated with this socket.
  29         int sock;
  30
  31         enum State { READING_REQUEST, SENDING_HEADER, SENDING_DATA };
  32         State state;
  33
  34         // The HTTP request, as sent by the client. If we are in READING_REQUEST,
  35         // this might not be finished.
  36         std::string request;
  37
  38         // What stream we're connecting to; parsed from <request>.
  39         // Not relevant for READING_REQUEST.
  40         std::string stream_id;
  41
  42         // The header we want to send. This is nominally a copy of Stream::header,
  43         // but since that might change on reconnects etc., we keep a local copy here.
  44         // Only relevant for SENDING_HEADER; blank otherwise.
  45         std::string header;
  46
  47         // Number of bytes we've sent of the header. Only relevant for SENDING_HEADER.
  48         size_t header_bytes_sent;
  49
  50         // Number of bytes we've sent of data. Only relevant for SENDING_DATA.
  51         size_t bytes_sent;
  52 };
  53
  54 struct Stream {
  55         Stream(const std::string &stream_id);
  56         ~Stream();
  57
  58         // Serialization/deserialization.
  59         Stream(const StreamProto &serialized);
  60         StreamProto serialize() const;
  61
  62         std::string stream_id;
  63
  64         // The HTTP response header, plus the video stream header (if any).
  65         std::string header;
  66
  67         // The stream data itself, stored in a circular buffer.
  68         char *data;
  69
  70         // How many bytes <data> contains. Can very well be larger than BACKLOG_SIZE,
  71         // since the buffer wraps.
  72         size_t data_size;
  73
  74 private:
  75         Stream(const Stream& other);
  76 };
  77
  78 class Server {
  79 public:
  80         Server();
  81         ~Server();
  82
  83         // Start a new thread that handles clients.
  84         void run();
  85
  86         // Stop the thread.
  87         void stop();
  88
  89         CubemapStateProto serialize() const;
  90
  91         void add_client(int sock);
  92         void add_client_from_serialized(const ClientProto &client);
  93
  94         void add_stream(const std::string &stream_id);
  95         void add_stream_from_serialized(const StreamProto &stream);
  96
  97         void set_header(const std::string &stream_id, const std::string &header);
  98         void add_data(const std::string &stream_id, const char *data, size_t bytes);
  99
 100 private:
 101         pthread_t worker_thread;
 102
 103         // All variables below this line are protected by the mutex.
 104         pthread_mutex_t mutex;
 105
 106         // If the thread should stop or not.
 107         bool should_stop;
 108
 109         // Map from stream ID to stream.
 110         std::map<std::string, Stream *> streams;
 111
 112         // Map from file descriptor to client.
 113         std::map<int, Client> clients;
 114
 115         // Used for epoll implementation (obviously).
 116         int epoll_fd;
 117         epoll_event events[EPOLL_MAX_EVENTS];
 118
 119         // Clients that are in SENDING_DATA, but that we don't listen on,
 120         // because we currently don't have any data for them.
 121         // See put_client_to_sleep() and wake_up_all_clients().
 122         std::vector<int> sleeping_clients;
 123
 124         // Recover the this pointer, and call do_work().
 125         static void *do_work_thunk(void *arg);
 126
 127         // The actual worker thread.
 128         void do_work();
 129
 130         void process_client(Client *client);
 131
 132         // Close a given client socket, and clean up after it.
 133         void close_client(Client *client);
 134
 135         // Parse the HTTP request.
 136         void parse_request(Client *client);
 137
 138         // Construct the HTTP header, and set the client into
 139         // the SENDING_HEADER state.
 140         void construct_header(Client *client);
 141
 142         // Put client to sleep, since there is no more data for it; we will on
 143         // longer listen on POLLOUT until we get more data. Also, it will be put
 144         // in the list of clients to wake up when we do.
 145         void put_client_to_sleep(Client *client);
 146
 147         // We have more data, so mark all clients that are sleeping as ready to go.
 148         void wake_up_all_clients();
 149
 150         // TODO: This function should probably die.
 151         Stream *find_stream(const std::string &stream_id);
 152 };
 153
 154 #endif  // !defined(_SERVER_H)