X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=httpinput.cpp;h=2fe68061b44c89e7b8a2811df8c5aff3369cb3bd;hp=f92ac13ca9ca010f1b2a8efb2d45cba68c40b12f;hb=2a2689e25c5906f66c5154803aae1d705dbf33af;hpb=3fd8650ccf3da3960a946d8ac9abc305aec399ce diff --git a/httpinput.cpp b/httpinput.cpp index f92ac13..2fe6806 100644 --- a/httpinput.cpp +++ b/httpinput.cpp @@ -1,4 +1,3 @@ -#include #include #include #include @@ -9,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -18,6 +18,7 @@ #include "httpinput.h" #include "log.h" #include "metacube.h" +#include "mutexlock.h" #include "parse.h" #include "serverpool.h" #include "state.pb.h" @@ -34,6 +35,11 @@ HTTPInput::HTTPInput(const string &url) has_metacube_header(false), sock(-1) { + pthread_mutex_init(&stats_mutex, NULL); + stats.url = url; + stats.bytes_received = 0; + stats.data_bytes_received = 0; + stats.connect_time = -1; } HTTPInput::HTTPInput(const InputProto &serialized) @@ -58,11 +64,26 @@ HTTPInput::HTTPInput(const InputProto &serialized) memcmp(http_header.data() + http_header.size() - 4, "\r\n\r\n", 4) == 0) { http_header.resize(http_header.size() - 2); } + + pthread_mutex_init(&stats_mutex, NULL); + stats.url = url; + stats.bytes_received = serialized.bytes_received(); + stats.data_bytes_received = serialized.data_bytes_received(); + if (serialized.has_connect_time()) { + stats.connect_time = serialized.connect_time(); + } else { + stats.connect_time = time(NULL); + } } void HTTPInput::close_socket() { - safe_close(sock); + if (sock != -1) { + safe_close(sock); + } + + MutexLock lock(&stats_mutex); + stats.connect_time = -1; } InputProto HTTPInput::serialize() const @@ -77,6 +98,9 @@ InputProto HTTPInput::serialize() const serialized.set_pending_data(string(pending_data.begin(), pending_data.end())); serialized.set_has_metacube_header(has_metacube_header); serialized.set_sock(sock); + serialized.set_bytes_received(stats.bytes_received); + serialized.set_data_bytes_received(stats.data_bytes_received); + serialized.set_connect_time(stats.connect_time); return serialized; } @@ -84,17 +108,16 @@ int HTTPInput::lookup_and_connect(const string &host, const string &port) { addrinfo *ai; int err = getaddrinfo(host.c_str(), port.c_str(), NULL, &ai); - if (err == -1) { + if (err != 0) { log(WARNING, "[%s] Lookup of '%s' failed (%s).", url.c_str(), host.c_str(), gai_strerror(err)); - freeaddrinfo(ai); return -1; } addrinfo *base_ai = ai; // Connect to everything in turn until we have a socket. - while (ai && !should_stop()) { + for ( ; ai && !should_stop(); ai = ai->ai_next) { int sock = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP); if (sock == -1) { // Could be e.g. EPROTONOSUPPORT. The show must go on. @@ -152,7 +175,6 @@ int HTTPInput::lookup_and_connect(const string &host, const string &port) } safe_close(sock); - ai = ai->ai_next; } // Give the last one as error. @@ -240,8 +262,8 @@ bool HTTPInput::parse_response(const std::string &request) http_header.append(it->first + ": " + it->second + "\r\n"); } - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], http_header, ""); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], http_header, ""); } return true; @@ -264,8 +286,9 @@ void HTTPInput::do_work() request_bytes_sent = 0; response.clear(); pending_data.clear(); - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], "", ""); + has_metacube_header = false; + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], "", ""); } { @@ -288,6 +311,9 @@ void HTTPInput::do_work() request = "GET " + path + " HTTP/1.0\r\nUser-Agent: cubemap\r\n\r\n"; request_bytes_sent = 0; } + + MutexLock lock(&stats_mutex); + stats.connect_time = time(NULL); } break; case SENDING_REQUEST: { @@ -351,7 +377,7 @@ void HTTPInput::do_work() char *ptr = static_cast( memmem(response.data(), response.size(), "\r\n\r\n", 4)); assert(ptr != NULL); - extra_data = string(ptr, &response[0] + response.size()); + extra_data = string(ptr + 4, &response[0] + response.size()); response.resize(ptr - response.data()); } @@ -385,7 +411,7 @@ void HTTPInput::do_work() if (ret == 0) { // This really shouldn't happen... - log(ERROR, "[%s] Socket unexpectedly closed while reading header", + log(ERROR, "[%s] Socket unexpectedly closed while reading data", url.c_str()); state = CLOSING_SOCKET; continue; @@ -419,6 +445,10 @@ void HTTPInput::do_work() void HTTPInput::process_data(char *ptr, size_t bytes) { pending_data.insert(pending_data.end(), ptr, ptr + bytes); + { + MutexLock mutex(&stats_mutex); + stats.bytes_received += bytes; + } for ( ;; ) { // If we don't have enough data (yet) for even the Metacube header, just return. @@ -457,21 +487,36 @@ void HTTPInput::process_data(char *ptr, size_t bytes) uint32_t size = ntohl(hdr->size); uint32_t flags = ntohl(hdr->flags); + if (size > 262144) { + log(WARNING, "[%s] Metacube block of %d bytes (flags=%x); corrupted header?", + url.c_str(), size, flags); + } + // See if we have the entire block. If not, wait for more data. if (pending_data.size() < sizeof(metacube_block_header) + size) { return; } - // Send this block on to the data. + // Send this block on to the servers. + { + MutexLock lock(&stats_mutex); + stats.data_bytes_received += size; + } char *inner_data = pending_data.data() + sizeof(metacube_block_header); if (flags & METACUBE_FLAGS_HEADER) { string header(inner_data, inner_data + size); - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], http_header, header); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], http_header, header); } - } else { - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->add_data(stream_ids[i], inner_data, size); + } else { + StreamStartSuitability suitable_for_stream_start; + if (flags & METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START) { + suitable_for_stream_start = NOT_SUITABLE_FOR_STREAM_START; + } else { + suitable_for_stream_start = SUITABLE_FOR_STREAM_START; + } + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->add_data(stream_indices[i], inner_data, size, suitable_for_stream_start); } } @@ -490,6 +535,18 @@ void HTTPInput::drop_pending_data(size_t num_bytes) } log(WARNING, "[%s] Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?", url.c_str(), (long long)num_bytes); + assert(pending_data.size() >= num_bytes); pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes); } +void HTTPInput::add_destination(int stream_index) +{ + stream_indices.push_back(stream_index); + servers->set_header(stream_index, http_header, ""); +} + +InputStats HTTPInput::get_stats() const +{ + MutexLock lock(&stats_mutex); + return stats; +}