X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=cubemap.cpp;h=692d48baf1d9e507d36926aa10a7f35f7d8847b1;hp=cbdab184dc7f5e3be2c098b65884ab4a1083c4b1;hb=f3e2d52ccf79454ee157917f9017263a8f4c50c3;hpb=a81b4cc30ee05292bdc3177680e5d7efbf09fa1f diff --git a/cubemap.cpp b/cubemap.cpp index cbdab18..692d48b 100644 --- a/cubemap.cpp +++ b/cubemap.cpp @@ -4,119 +4,117 @@ #include #include #include +#include +#include +#include +#include +#include +#include #include +#include +#include + #include "metacube.h" +#include "server.h" +#include "input.h" + +#define NUM_SERVERS 4 +#define STREAM_ID "stream" +#define STREAM_URL "http://gruessi.zrh.sesse.net:4013/" +#define PORT 9094 using namespace std; -class Input { -public: - Input(); - void curl_callback(char *ptr, size_t bytes); +Server *servers = NULL; + +int create_server_socket(int port) +{ + int server_sock = socket(PF_INET6, SOCK_STREAM, IPPROTO_TCP); + if (server_sock == -1) { + perror("socket"); + exit(1); + } -private: - void process_block(const char *data, uint32_t size, uint32_t flags); - void drop_pending_data(size_t num_bytes); + int one = 1; + if (setsockopt(server_sock, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) == -1) { + perror("setsockopt(SO_REUSEADDR)"); + exit(1); + } - // Data we have received but not fully processed yet. - vector pending_data; + // We want dual-stack sockets. (Sorry, OpenBSD and Windows XP...) + int zero = 0; + if (setsockopt(server_sock, IPPROTO_IPV6, IPV6_V6ONLY, &zero, sizeof(zero)) == -1) { + perror("setsockopt(IPV6_V6ONLY)"); + exit(1); + } - // If starts with a Metacube header, - // this is true. - bool has_metacube_header; -}; + sockaddr_in6 addr; + memset(&addr, 0, sizeof(addr)); + addr.sin6_family = AF_INET6; + addr.sin6_port = htons(port); -Input::Input() - : has_metacube_header(false) -{ + if (bind(server_sock, reinterpret_cast(&addr), sizeof(addr)) == -1) { + perror("bind"); + exit(1); + } + + if (listen(server_sock, 128) == -1) { + perror("listen"); + exit(1); + } + + return server_sock; } - -void Input::curl_callback(char *ptr, size_t bytes) -{ - pending_data.insert(pending_data.end(), ptr, ptr + bytes); +void *acceptor_thread_run(void *arg) +{ + int server_sock = int(intptr_t(arg)); + int num_accepted = 0; for ( ;; ) { - // If we don't have enough data (yet) for even the Metacube header, just return. - if (pending_data.size() < sizeof(metacube_block_header)) { - return; - } + sockaddr_in6 addr; + socklen_t addrlen = sizeof(addr); - // Make sure we have the Metacube sync header at the start. - // We may need to skip over junk data (it _should_ not happen, though). - if (!has_metacube_header) { - char *ptr = static_cast( - memmem(pending_data.data(), pending_data.size(), - METACUBE_SYNC, strlen(METACUBE_SYNC))); - if (ptr == NULL) { - // OK, so we didn't find the sync marker. We know then that - // we do not have the _full_ marker in the buffer, but we - // could have N-1 bytes. Drop everything before that, - // and then give up. - drop_pending_data(pending_data.size() - (strlen(METACUBE_SYNC) - 1)); - return; - } else { - // Yay, we found the header. Drop everything (if anything) before it. - drop_pending_data(ptr - pending_data.data()); - has_metacube_header = true; - - // Re-check that we have the entire header; we could have dropped data. - if (pending_data.size() < sizeof(metacube_block_header)) { - return; - } - } + // Get a new socket. + int sock = accept(server_sock, reinterpret_cast(&addr), &addrlen); + if (sock == -1 && errno == EINTR) { + continue; } - - // Now it's safe to read the header. - metacube_block_header *hdr = reinterpret_cast(pending_data.data()); - assert(memcmp(hdr->sync, METACUBE_SYNC, sizeof(hdr->sync)) == 0); - uint32_t size = ntohl(hdr->size); - uint32_t flags = ntohl(hdr->flags); - - // See if we have the entire block. If not, wait for more data. - if (pending_data.size() < sizeof(metacube_block_header) + size) { - return; + if (sock == -1) { + perror("accept"); + exit(1); } - process_block(pending_data.data(), size, flags); + // Set the socket as nonblocking. + int one = 1; + if (ioctl(sock, FIONBIO, &one) == -1) { + perror("FIONBIO"); + exit(1); + } - // Consume this block. This isn't the most efficient way of dealing with things - // should we have many blocks, but these routines don't need to be too efficient - // anyway. - pending_data.erase(pending_data.begin(), pending_data.begin() + sizeof(metacube_block_header) + size); + // Pick a server, round-robin, and hand over the socket to it. + servers[num_accepted % NUM_SERVERS].add_client(sock); + ++num_accepted; } } - -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); -} - -void Input::drop_pending_data(size_t num_bytes) + +int main(int argc, char **argv) { - if (num_bytes == 0) { - return; + servers = new Server[NUM_SERVERS]; + for (int i = 0; i < NUM_SERVERS; ++i) { + servers[i].add_stream(STREAM_ID); + servers[i].run(); } - fprintf(stderr, "Warning: Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?\n", - (long long)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 server_sock = create_server_socket(PORT); -int main(int argc, char **argv) -{ - Input input; - CURL *curl = curl_easy_init(); - curl_easy_setopt(curl, CURLOPT_URL, "http://gruessi.zrh.sesse.net:4013/"); - // curl_easy_setopt(curl, CURLOPT_URL, "http://www.google.com/"); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_callback); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &input); - curl_easy_perform(curl); + pthread_t acceptor_thread; + pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast(server_sock)); + + Input input(STREAM_ID); + input.run(STREAM_URL); + + for (int i = 0; i < NUM_SERVERS; ++i) { + servers[i].stop(); + } + delete[] servers; }