X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=udpinput.cpp;h=001ef5f6d3db7a5fbf89ec279482b27241eb1a16;hp=081f631c47e114ba42f7ff46699667d8d1d31c5e;hb=07569f8e011cd9b064c64bef1ce56f77bf7ddf53;hpb=d1ad2cf60ca694f742e2061a596bc27c7e58f5c8 diff --git a/udpinput.cpp b/udpinput.cpp index 081f631..001ef5f 100644 --- a/udpinput.cpp +++ b/udpinput.cpp @@ -1,124 +1,237 @@ #include #include #include -#include +#include #include #include +#include #include #include #include "acceptor.h" #include "log.h" +#include "mutexlock.h" #include "serverpool.h" #include "state.pb.h" +#include "stream.h" #include "udpinput.h" +#include "util.h" #include "version.h" using namespace std; extern ServerPool *servers; -UDPInput::UDPInput(const string &stream_id, const string &url) - : stream_id(stream_id), - url(url), +namespace { + +// Similar to parse_hostport(), but only parses the IP address, +// and does not use mapped-v4 addresses, since multicast seems +// to not like that too much. +bool parse_ip_address(const string &ip, sockaddr_storage *addr) +{ + memset(addr, 0, sizeof(*addr)); + + assert(!ip.empty()); + if (ip[0] == '[') { + sockaddr_in6 *addr6 = (sockaddr_in6 *)addr; + addr6->sin6_family = AF_INET6; + if (ip[ip.size() - 1] != ']') { + log(ERROR, "address '%s' is malformed; must be either [ipv6addr] or ipv4addr", + ip.c_str()); + return false; + } + if (inet_pton(AF_INET6, ip.c_str(), &addr6->sin6_addr) != 1) { + log(ERROR, "'%s' is not a valid IPv6 address"); + return false; + } + } else { + sockaddr_in *addr4 = (sockaddr_in *)addr; + addr4->sin_family = AF_INET; + if (inet_pton(AF_INET, ip.c_str(), &addr4->sin_addr) != 1) { + log(ERROR, "'%s' is not a valid IPv4 address"); + return false; + } + } + + return true; +} + +bool maybe_join_multicast_group(int sock, const string &group, const string &source) +{ + if (group.empty()) { + // Not multicast. + return true; + } + + // Join the given multicast group (ASM or SSM). + // TODO: Also support sources apart from multicast groups, + // e.g. udp://[::1]:1234 for only receiving from localhost. + if (!source.empty()) { + // Single-Source Multicast (SSM). + group_source_req gsr; + memset(&gsr, 0, sizeof(gsr)); + if (!parse_ip_address(group, &gsr.gsr_group)) { + return false; + } + if (!parse_ip_address(source, &gsr.gsr_source)) { + return false; + } + int level = (gsr.gsr_group.ss_family == AF_INET) ? SOL_IP : SOL_IPV6; + if (setsockopt(sock, level, MCAST_JOIN_SOURCE_GROUP, &gsr, sizeof(gsr)) == -1) { + log_perror("setsockopt(MCAST_JOIN_SOURCE_GROUP)"); + return false; + } + } else { + // Any-Source Multicast (ASM). + group_req gr; + memset(&gr, 0, sizeof(gr)); + if (!parse_ip_address(group, &gr.gr_group)) { + return false; + } + int level = (gr.gr_group.ss_family == AF_INET) ? SOL_IP : SOL_IPV6; + if (setsockopt(sock, level, MCAST_JOIN_GROUP, &gr, sizeof(gr)) == -1) { + log_perror("setsockopt(MCAST_JOIN_GROUP)"); + return false; + } + } + + return true; +} + +} // namespace + +UDPInput::UDPInput(const string &url) + : url(url), sock(-1) { // Should be verified by the caller. string protocol; - bool ok = parse_url(url, &protocol, &host, &port, &path); + bool ok = parse_url(url, &protocol, &user, &host, &port, &path); assert(ok); construct_header(); + + pthread_mutex_init(&stats_mutex, NULL); + stats.url = url; + stats.bytes_received = 0; + stats.data_bytes_received = 0; + stats.connect_time = time(NULL); } UDPInput::UDPInput(const InputProto &serialized) - : stream_id(serialized.stream_id()), - url(serialized.url()), + : url(serialized.url()), sock(serialized.sock()) { // Should be verified by the caller. string protocol; - bool ok = parse_url(url, &protocol, &host, &port, &path); + bool ok = parse_url(url, &protocol, &user, &host, &port, &path); assert(ok); construct_header(); + + 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); + } } InputProto UDPInput::serialize() const { InputProto serialized; - serialized.set_stream_id(stream_id); serialized.set_url(url); 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; } void UDPInput::close_socket() { - int ret; - do { - ret = close(sock); - } while (ret == -1 && errno == EINTR); - - if (ret == -1) { - perror("close()"); - } - + safe_close(sock); sock = -1; } void UDPInput::construct_header() { - string header = + http_header = "HTTP/1.0 200 OK\r\n" "Content-type: application/octet-stream\r\n" "Cache-control: no-cache\r\n" "Server: " SERVER_IDENTIFICATION "\r\n" - "\r\n"; - servers->set_header(stream_id, header); + "Connection: close\r\n"; +} + +void UDPInput::add_destination(int stream_index) +{ + stream_indices.push_back(stream_index); + servers->set_header(stream_index, http_header, ""); } void UDPInput::do_work() { - while (!should_stop) { + while (!should_stop()) { if (sock == -1) { int port_num = atoi(port.c_str()); - sock = create_server_socket(port_num, UDP_SOCKET); + sockaddr_in6 addr = create_any_address(port_num); + sock = create_server_socket(addr, UDP_SOCKET); if (sock == -1) { - log(WARNING, "UDP socket creation failed. Waiting 0.2 seconds and trying again..."); + log(WARNING, "[%s] UDP socket creation failed. Waiting 0.2 seconds and trying again...", + url.c_str()); usleep(200000); continue; } - } - // Since we are non-blocking, we need to wait for the right state first. - // Wait up to 50 ms, then check should_stop. - pollfd pfd; - pfd.fd = sock; - pfd.events = POLLIN; + // The syntax udp://source@group (abusing the username field + // to store the sender in SSM) seems to be a VLC invention. + // We mimic it. + if (!maybe_join_multicast_group(sock, host, user)) { + log(WARNING, "[%s] Multicast join failed. Waiting 0.2 seconds and trying again...", + url.c_str()); + safe_close(sock); + sock = -1; + usleep(200000); + continue; + } + } - int nfds = poll(&pfd, 1, 50); - if (nfds == 0 || (nfds == -1 && errno == EINTR)) { + // Wait for a packet, or a wakeup. + bool activity = wait_for_activity(sock, POLLIN, NULL); + if (!activity) { + // Most likely, should_stop was set. continue; } - if (nfds == -1) { - perror("poll"); - close_socket(); - continue; - } - char buf[4096]; int ret; do { - ret = recv(sock, buf, sizeof(buf), 0); + ret = recv(sock, packet_buf, sizeof(packet_buf), 0); } while (ret == -1 && errno == EINTR); if (ret <= 0) { - perror("recv"); + log_perror("recv"); close_socket(); continue; } + + { + MutexLock lock(&stats_mutex); + stats.bytes_received += ret; + stats.data_bytes_received += ret; + } - servers->add_data(stream_id, buf, ret); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->add_data(stream_indices[i], packet_buf, ret, SUITABLE_FOR_STREAM_START); + } } } + +InputStats UDPInput::get_stats() const +{ + MutexLock lock(&stats_mutex); + return stats; +}