From: Steinar H. Gunderson Date: Sat, 6 Apr 2013 14:01:02 +0000 (+0200) Subject: Piece a lot more stuff together. X-Git-Tag: 1.0.0~223 X-Git-Url: https://git.sesse.net/?p=cubemap;a=commitdiff_plain;h=2bd0c02ac79dc8bc0da62b27d6b1e9f3cea525c6;hp=c555641af6f4021d744a50abd5084a174614dd54 Piece a lot more stuff together. --- diff --git a/cubemap.cpp b/cubemap.cpp index 9f5e42e..643f1a8 100644 --- a/cubemap.cpp +++ b/cubemap.cpp @@ -7,6 +7,8 @@ #include #include #include +#include +#include #include #include #include @@ -81,15 +83,22 @@ public: // Start a new thread that handles clients. void run(); - void add_socket(int server_sock); + void add_client(int sock); void add_stream(const string &stream_id); void set_header(const string &stream_id, const string &header); void add_data(const string &stream_id, const char *data, size_t bytes); private: pthread_mutex_t mutex; + + // Map from stream ID to stream. map streams; + // Map from file descriptor to client. + map clients; + + int epoll_fd; + // Recover the this pointer, and call do_work(). static void *do_work_thunk(void *arg); @@ -97,9 +106,17 @@ private: void do_work(); }; +Server *servers = NULL; + Server::Server() { pthread_mutex_init(&mutex, NULL); + + epoll_fd = epoll_create(1024); // Size argument is ignored. + if (epoll_fd == -1) { + perror("epoll_fd"); + exit(1); + } } void Server::run() @@ -118,20 +135,71 @@ void *Server::do_work_thunk(void *arg) void Server::do_work() { for ( ;; ) { + MutexLock lock(&mutex); printf("server thread running\n"); sleep(1); } } + +void Server::add_client(int sock) +{ + MutexLock lock(&mutex); + Client new_client; + new_client.state = Client::READING_REQUEST; + new_client.header_bytes_sent = 0; + new_client.bytes_sent = 0; + + clients.insert(make_pair(sock, new_client)); + + // Start listening on data from this socket. + epoll_event ev; + ev.events = EPOLLIN; + ev.data.fd = sock; + if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_ADD)"); + exit(1); + } +} + +void Server::add_stream(const string &stream_id) +{ + // TODO +} + +void Server::set_header(const string &stream_id, const string &header) +{ + // TODO + printf("got header! %lu bytes\n", header.size()); +} + +void Server::add_data(const string &stream_id, const char *data, size_t bytes) +{ + // TODO +} class Input { public: - Input(); - void curl_callback(char *ptr, size_t bytes); + Input(const string &stream_id); + + // Connect to the given URL and start streaming. + // WARNING: Currently this blocks; it does not run in a separate thread! + void run(const string &url); private: + // Recovers the this pointer and calls curl_callback(). + static size_t curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata); + + // Stores the given data, looks for Metacube blocks (skipping data if needed), + // and calls process_block() for each one. + void curl_callback(char *ptr, size_t bytes); void process_block(const char *data, uint32_t size, uint32_t flags); + + // Drops bytes from the head of , + // and outputs a warning. void drop_pending_data(size_t num_bytes); + string stream_id; + // Data we have received but not fully processed yet. vector pending_data; @@ -140,10 +208,28 @@ private: bool has_metacube_header; }; -Input::Input() - : has_metacube_header(false) +Input::Input(const string &stream_id) + : stream_id(stream_id), + has_metacube_header(false) { } + +void Input::run(const string &url) +{ + CURL *curl = curl_easy_init(); + curl_easy_setopt(curl, CURLOPT_URL, STREAM_URL); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &Input::curl_callback_thunk); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, this); + curl_easy_perform(curl); +} + +size_t Input::curl_callback_thunk(char *ptr, size_t size, size_t nmemb, void *userdata) +{ + Input *input = static_cast(userdata); + size_t bytes = size * nmemb; + input->curl_callback(ptr, bytes); + return bytes; +} void Input::curl_callback(char *ptr, size_t bytes) { @@ -201,9 +287,17 @@ void Input::curl_callback(char *ptr, size_t bytes) } void Input::process_block(const char *data, uint32_t size, uint32_t flags) -{ - // TODO: treat it right here - printf("Block: %d bytes, flags=0x%x\n", size, flags); +{ + if (flags & METACUBE_FLAGS_HEADER) { + string header(data, data + size); + for (int i = 0; i < NUM_SERVERS; ++i) { + servers[i].set_header(stream_id, header); + } + } else { + for (int i = 0; i < NUM_SERVERS; ++i) { + servers[i].add_data(stream_id, data, size); + } + } } void Input::drop_pending_data(size_t num_bytes) @@ -216,14 +310,6 @@ void Input::drop_pending_data(size_t num_bytes) pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes); } -size_t curl_callback(char *ptr, size_t size, size_t nmemb, void *userdata) -{ - Input *input = static_cast(userdata); - size_t bytes = size * nmemb; - input->curl_callback(ptr, bytes); - return bytes; -} - int create_server_socket(int port) { int server_sock = socket(PF_INET6, SOCK_STREAM, IPPROTO_TCP); @@ -260,10 +346,12 @@ int create_server_socket(int port) void *acceptor_thread_run(void *arg) { int server_sock = int(intptr_t(arg)); + int num_accepted = 0; for ( ;; ) { sockaddr_in6 addr; socklen_t addrlen = sizeof(addr); + // Get a new socket. int sock = accept(server_sock, reinterpret_cast(&addr), &addrlen); if (sock == -1 && errno == EINTR) { continue; @@ -273,16 +361,24 @@ void *acceptor_thread_run(void *arg) exit(1); } - printf("got a socket yaaaay\n"); + // Set the socket as nonblocking. + int one = 1; + if (ioctl(sock, FIONBIO, &one) == -1) { + perror("FIONBIO"); + exit(1); + } + + // Pick a server, round-robin, and hand over the socket to it. + servers[num_accepted % NUM_SERVERS].add_client(sock); + ++num_accepted; } } -Server *servers = NULL; - int main(int argc, char **argv) { servers = new Server[NUM_SERVERS]; for (int i = 0; i < NUM_SERVERS; ++i) { + servers[i].add_stream(STREAM_ID); servers[i].run(); } @@ -291,10 +387,6 @@ int main(int argc, char **argv) pthread_t acceptor_thread; pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast(server_sock)); - Input input; - CURL *curl = curl_easy_init(); - curl_easy_setopt(curl, CURLOPT_URL, STREAM_URL); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_callback); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &input); - curl_easy_perform(curl); + Input input(STREAM_ID); + input.run(STREAM_URL); }