hd-rum-translator.cpp File Reference

#include <atomic>
#include <cassert>
#include <cctype>
#include <cinttypes>
#include <climits>
#include <csignal>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <getopt.h>
#include <memory>
#include <pthread.h>
#include <stdexcept>
#include <string>
#include <thread>
#include <unistd.h>
#include <utility>
#include <vector>
#include "control_socket.h"
#include "debug.h"
#include "hd-rum-translator/hd-rum-decompress.h"
#include "hd-rum-translator/hd-rum-recompress.h"
#include "host.h"
#include "lib_common.h"
#include "messaging.h"
#include "module.h"
#include "rtp/net_udp.h"
#include "tv.h"
#include "ug_runtime_error.hpp"
#include "utils/color_out.h"
#include "utils/misc.h"
#include "utils/net.h"

Include dependency graph for hd-rum-translator.cpp:

Go to the source code of this file.

Classes
struct	cmdline_parameters
struct	Conf_participant
struct	hd_rum_translator_state
struct	host_opts
struct	item
class	Participant_manager
struct	replica

Macros
#define	EXIT(retval)
#define	MAX_PKT_SIZE   10000
#define	MOD_NAME   "[hd-rum-trans] "
#define	prefix_matches(x, y)
#define	REPLICA_MAGIC   0xd2ff3323
#define	SIZE   MAX_PKT_SIZE

Functions
static struct response *	change_replica_type (struct hd_rum_translator_state *s, struct module *mod, struct message *msg, int index)
static int	create_output_port (struct hd_rum_translator_state *s, const char *addr, int rx_port, int tx_port, int bufsize, struct common_opts *common, const char *compression, const char *fec, int bitrate, bool use_server_sock=false)
static char *	get_replica_mod_name (const char *addr, uint16_t tx_port)
static void	hd_rum_translator_deinit (struct hd_rum_translator_state *s)
static void	hd_rum_translator_should_exit_callback (void *arg)
int	main (int argc, char **argv)
static int	parse_fmt (int argc, char **argv, struct cmdline_parameters *parsed) noexcept(false)
static int	parse_global_opts (int argc, char **argv, struct cmdline_parameters *parsed) noexcept(false)
static int	parse_size (const char *sz_str) noexcept(false)
	unit_evaluate() is similar but uses SI prefixes
static void	qdestroy (struct item *queue)
static struct item *	qinit (int qsize)
static void	signal_handler (int signum)
static bool	sockaddr_equal (struct sockaddr_storage *a, struct sockaddr_storage *b)
static void	usage (const char *progname)
static void *	writer (void *arg)

Detailed Description

Author

Martin Pulec pulec.nosp@m.@ces.nosp@m.net.c.nosp@m.z

Martin Piatka piatk.nosp@m.a@ce.nosp@m.snet..nosp@m.cz

Main part of transcoding reflector. This component provides a runtime for the reflector. Componets are following:

• legacy packet reflector (defined in this file) for backward compatiblity. It handles those recipient that doesn't need transcoding.
• decompressor - decompresses the stream if there are some host that need transcoding
• recompressor - for every transcoded host, there is a recompressor that compresses and sends frame to receiver

Definition in file hd-rum-translator.cpp.

Macro Definition Documentation

EXIT

#define EXIT

(

retval

)

Value:

    { \
            exit_uv(0); \
            hd_rum_translator_deinit(&state); \
            if (sock_in != nullptr) \
                    udp_exit(sock_in); \
            common_cleanup(init); \
            return retval; \
    }

Definition at line 973 of file hd-rum-translator.cpp.

#define EXIT(retval) \
    { \
            exit_uv(0); \
            hd_rum_translator_deinit(&state); \
            if (sock_in != nullptr) \
                    udp_exit(sock_in); \
            common_cleanup(init); \
            return retval; \
    }

Referenced by main().

MAX_PKT_SIZE

#define MAX_PKT_SIZE   10000

Definition at line 221 of file hd-rum-translator.cpp.

MOD_NAME

#define MOD_NAME   "[hd-rum-trans] "

Definition at line 89 of file hd-rum-translator.cpp.

prefix_matches

#define prefix_matches	(		x,
			y )

Value:

strncasecmp(x, y, strlen(y)) == 0

Definition at line 282 of file hd-rum-translator.cpp.

Referenced by change_replica_type().

REPLICA_MAGIC

#define REPLICA_MAGIC   0xd2ff3323

Definition at line 91 of file hd-rum-translator.cpp.

Referenced by replica::replica(), and replica::~replica().

SIZE

#define SIZE   MAX_PKT_SIZE

Definition at line 232 of file hd-rum-translator.cpp.

Referenced by main(), qinit(), and writer().

Function Documentation

change_replica_type()

static struct response * change_replica_type ( struct hd_rum_translator_state * s, struct module * mod, struct message * msg, int index )

static

Definition at line 283 of file hd-rum-translator.cpp.

{
    struct replica *r = (struct replica *) mod->priv_data;
 
    struct msg_universal *data = (struct msg_universal *) msg;
 
    if (strcasecmp(data->text, "sock") == 0) {
        r->type = replica::type_t::USE_SOCK;
        log_msg(LOG_LEVEL_NOTICE, "Output port %d is now forwarding.\n", index);
    } else if (strcasecmp(data->text, "recompress") == 0) {
        r->type = replica::type_t::RECOMPRESS;
        log_msg(LOG_LEVEL_NOTICE, "Output port %d is now transcoding.\n", index);
    } else if (prefix_matches(data->text, "compress ")) {
        if(recompress_port_change_compress(s->recompress, index, data->text + strlen("compress "))){
            log_msg(LOG_LEVEL_NOTICE, "Output port %d compression changed.\n", index);
        } else {
            log_msg(LOG_LEVEL_ERROR, "Failed to change port %d compression. Port removed.\n", index);
        }
    } else {
        fprintf(stderr, "Unknown replica type \"%s\"\n", data->text);
        return new_response(RESPONSE_BAD_REQUEST, NULL);
    }
 
    recompress_port_set_active(s->recompress, index,
            r->type == replica::type_t::RECOMPRESS);
 
    return new_response(RESPONSE_OK, NULL);
}

References LOG_LEVEL_ERROR, LOG_LEVEL_NOTICE, log_msg(), replica::mod, new_response(), prefix_matches, module::priv_data, hd_rum_translator_state::recompress, recompress_port_change_compress(), recompress_port_set_active(), RESPONSE_BAD_REQUEST, RESPONSE_OK, msg_universal::text, and replica::type.

Referenced by writer().

Here is the call graph for this function:

create_output_port()

static int create_output_port ( struct hd_rum_translator_state * s, const char * addr, int rx_port, int tx_port, int bufsize, struct common_opts * common, const char * compression, const char * fec, int bitrate, bool use_server_sock = false )

static

Definition at line 325 of file hd-rum-translator.cpp.

{
        struct replica *rep;
        try {
            rep = new replica(addr, rx_port, tx_port, bufsize, &s->mod,
                              common->force_ip_version);
            if(use_server_sock){
                    rep->sock = s->server_socket;
            }
        } catch (string const & s) {
            fputs(s.c_str(), stderr);
            const char *err_msg = "cannot create output port (wrong address?)";
            log_msg(LOG_LEVEL_ERROR, "%s\n", err_msg);
            return -1;
        }
        s->replicas.push_back(rep);
 
        struct common_opts com_opts = *common;
        com_opts.parent = &rep->mod;
        rep->type = compression ? replica::type_t::RECOMPRESS : replica::type_t::USE_SOCK;
        int idx = recompress_add_port(s->recompress,
                addr, compression ? compression : "none",
                0, tx_port, &com_opts, fec, bitrate);
        if (idx < 0) {
            fprintf(stderr, "Initializing output port '%s' compression failed!\n", addr);
 
            delete s->replicas.back();
            s->replicas.pop_back();
 
            return -1;
        }
 
        assert((unsigned) idx == s->replicas.size() - 1);
        recompress_port_set_active(s->recompress, idx, compression != nullptr);
 
        return idx;
}

References replica::replica(), common_opts::force_ip_version, LOG_LEVEL_ERROR, log_msg(), hd_rum_translator_state::mod, replica::mod, common_opts::parent, hd_rum_translator_state::recompress, recompress_add_port(), recompress_port_set_active(), hd_rum_translator_state::replicas, hd_rum_translator_state::server_socket, replica::sock, and replica::type.

Referenced by main(), and writer().

Here is the call graph for this function:

get_replica_mod_name()

static char * get_replica_mod_name ( const char * addr, uint16_t tx_port )

static

Definition at line 93 of file hd-rum-translator.cpp.

                                                                     {
        const size_t len = strlen(addr) + 2 /* [ ] for IPv6 addr */ + 6 /* port (including ':') */ + 1 /* '\0' */;
        char *name = (char *) malloc(len);
        bool is_ipv6 = strchr(addr, ':') != NULL;
        bool add_bracket = is_ipv6 && addr[0] != '[';
        snprintf(name, len, "%s%s%s:%" PRIu16, add_bracket ? "[" : "", addr, add_bracket ? "]" : "", tx_port);
 
        return name;
}

References len, and name.

Referenced by replica::replica(), and Participant_manager::tick().

hd_rum_translator_deinit()

static void hd_rum_translator_deinit ( struct hd_rum_translator_state * s )

static

Definition at line 822 of file hd-rum-translator.cpp.

                                                                        {
    if(s->decompress) {
        hd_rum_decompress_done(s->decompress);
    }
 
    if(s->recompress)
            recompress_done(s->recompress);
 
    for (unsigned int i = 0; i < s->replicas.size(); i++) {
        delete s->replicas[i];
    }
 
    control_done(s->control_state);
 
    qdestroy(s->queue);
}

References control_done(), hd_rum_translator_state::control_state, hd_rum_translator_state::decompress, hd_rum_decompress_done(), qdestroy(), hd_rum_translator_state::queue, hd_rum_translator_state::recompress, recompress_done(), and hd_rum_translator_state::replicas.

Referenced by main().

Here is the call graph for this function:

hd_rum_translator_should_exit_callback()

static void hd_rum_translator_should_exit_callback ( void * arg )

static

Definition at line 968 of file hd-rum-translator.cpp.

                                                              {
    volatile auto *should_exit = (volatile bool *) arg;
    *should_exit = true;
}

Referenced by main().

main()

int main	(	int	argc,
		char **	argv )

Definition at line 983 of file hd-rum-translator.cpp.

{
    struct init_data *init;
    struct hd_rum_translator_state state;
 
    int qsize;
    socket_udp *sock_in = nullptr;
    pthread_t thread;
    int err = 0;
    int i;
    struct cmdline_parameters params = {};
 
    if ((init = common_preinit(argc, argv)) == nullptr) {
        EXIT(EXIT_FAILURE);
    }
 
    print_version();
    printf("\n");
 
#ifndef _WIN32
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = signal_handler;
    sigemptyset(&sa.sa_mask);
 
    sigaction(SIGINT, &sa, NULL);
    sigaction(SIGTERM, &sa, NULL);
    sigaction(SIGHUP, &sa, NULL);
    sigaction(SIGABRT, &sa, NULL);
    sigaction(SIGPIPE, &sa, NULL);
    signal(SIGQUIT, crash_signal_handler);
    signal(SIGALRM, hang_signal_handler);
#else
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
    signal(SIGABRT, signal_handler);
#endif
    signal(SIGABRT, crash_signal_handler);
    signal(SIGFPE, crash_signal_handler);
    signal(SIGILL, crash_signal_handler);
    signal(SIGSEGV, crash_signal_handler);
 
    int ret = -1;
    try {
        ret = parse_fmt(argc, argv, &params);
    } catch (invalid_argument &e) {
        if (!invalid_arg_is_numeric(e.what())) {
            throw;
        }
        MSG(ERROR, "Non-numeric value passed to option "
                   "expecting a number!\n");
    } catch (ug_runtime_error &e) {
        MSG(FATAL, "%s\n", e.what());
    }
    if (ret != 0) {
        EXIT(ret < 0 ? EXIT_FAILURE : EXIT_SUCCESS);
    }
 
    if (params.log_level != -1) {
        log_level = params.log_level;
    }
 
    state.bufsize = params.bufsize;
 
    qsize = state.bufsize / 8000;
 
    printf("using UDP send and receive buffer size of %d bytes\n", state.bufsize);
 
    state.qhead = state.qtail = state.queue = qinit(qsize);
    if (!state.qhead) {
        EXIT(EXIT_FAILURE);
    }
 
    /* input socket */
    if ((sock_in = udp_init_if("localhost", NULL, params.port, 0, 255, false, false)) == NULL) {
        perror("input socket");
        EXIT(EXIT_FAILURE);
    }
 
    if (!udp_set_recv_buf(sock_in, state.bufsize)) {
        fprintf(stderr, "Cannot set recv buffer to %sB!\n",
                format_in_si_units(state.bufsize));
    }
 
    printf("listening on *:%d\n", udp_get_udp_rx_port(sock_in));
 
    if (params.control_port != -1) {
        if (control_init(params.control_port, params.control_connection_type, &state.control_state, &state.mod, 0) != 0) {
            fprintf(stderr, "Warning: Unable to create remote control.\n");
            EXIT(EXIT_FAIL_CONTROL_SOCK);
        }
        control_start(state.control_state);
    }
 
        //one shared recompressor, which manages compressions and sends to all hosts
        state.recompress = recompress_init(&state.mod);
    if(!state.recompress) {
        EXIT(EXIT_FAIL_COMPRESS);
    }
 
    // we need only one shared receiver decompressor for all recompressing streams
    state.decompress = hd_rum_decompress_init(&state.mod, params.out_conf,
                        params.capture_filter, state.recompress);
    if(!state.decompress) {
        EXIT(EXIT_FAIL_DECODER);
    }
 
    if(params.server_port > 0){
            state.server_socket = std::shared_ptr<socket_udp>(udp_init("localhost", params.server_port, 0, 255, 0, false), udp_exit);
            if (!state.server_socket) {
                EXIT(EXIT_FAILURE);
            }
    }
 
    if(params.out_conf.mode == CONFERENCE && !params.conference_compression){
            params.conference_compression = "libavcodec";
    }
 
    for (i = 0; i < params.host_count; i++) {
        int tx_port = params.port; // use default rx port by default
        int rx_port = params.hosts[i].rx_port;
        if (params.hosts[i].tx_port) {
            tx_port = params.hosts[i].tx_port;
        }
 
        auto& h = params.hosts[i];
 
        if(!h.compression && params.out_conf.mode == CONFERENCE){
            h.compression = params.conference_compression;
        }
 
        int idx = create_output_port(&state,
                h.addr, rx_port, tx_port, state.bufsize, &h.common_opts,
                h.compression, h.compression ? h.fec : nullptr, h.bitrate);
        if(idx < 0) {
            EXIT(EXIT_FAILURE);
        }
    }
 
    if (pthread_create(&thread, NULL, writer, (void *) &state)) {
        fprintf(stderr, "cannot create writer thread\n");
        EXIT(2);
    }
 
    uint64_t received_data = 0;
    struct timeval t0;
    gettimeofday(&t0, NULL);
 
    unsigned long long int last_data = 0ull;
 
    Participant_manager participant_mgr(state.mod, params.conference_compression);
 
    if(state.server_socket){
            participant_mgr.run_async(state.server_socket);
    }
 
    volatile bool should_exit = false;
    register_should_exit_callback(&state.mod, hd_rum_translator_should_exit_callback, const_cast<bool *>(&should_exit));
    /* main loop */
    while (!should_exit) {
        while (state.qtail->next != state.qhead && !should_exit) {
            struct timeval timeout = { 1, 0 };
 
            struct sockaddr_storage sin = {};
            socklen_t addrlen = sizeof(sin);
            state.qtail->size = udp_recvfrom_timeout(sock_in, state.qtail->buf, SIZE, &timeout, (sockaddr *) &sin, &addrlen);
            if(state.qtail->size <= 0)
                break;
 
            struct timeval t;
            gettimeofday(&t, NULL);
 
            if(params.out_conf.mode == CONFERENCE){
                    participant_mgr.tick(sin, addrlen);
            }
 
            received_data += state.qtail->size;
 
            state.qtail = state.qtail->next;
 
            pthread_mutex_lock(&state.qempty_mtx);
            state.qempty = 0;
            pthread_cond_signal(&state.qempty_cond);
            pthread_mutex_unlock(&state.qempty_mtx);
 
            double seconds = tv_diff(t, t0);
            if (seconds > 5.0) {
                unsigned long long int cur_data = (received_data - last_data);
                unsigned long long int bps = cur_data / seconds;
                char tim_str[20];
                time_t tim = time(NULL);
                struct tm *tmp = localtime(&tim);
                if (tmp) {
                    strftime(tim_str, sizeof(tim_str), "%F %T", tmp);
                }
                log_msg(LOG_LEVEL_INFO, "[%s] Received %llu bytes in %g seconds = %sbps\n", tim_str, cur_data, seconds, format_in_si_units(bps * 8));
                t0 = t;
                last_data = received_data;
            }
        }
 
        if (state.qtail->size <= 0)
            continue;
 
        pthread_mutex_lock(&state.qfull_mtx);
        if (state.qfull)
            pthread_cond_wait(&state.qfull_cond, &state.qfull_mtx);
        state.qfull = 1;
        pthread_mutex_unlock(&state.qfull_mtx);
    }
 
    if (state.qtail->size < 0 && !should_exit) {
        printf("read: %s\n", strerror(err));
        EXIT(2);
    }
 
    // pass poisoned pill to the worker
    state.qtail->size = 0;
    state.qtail = state.qtail->next;
 
    pthread_mutex_lock(&state.qempty_mtx);
    state.qempty = 0;
    pthread_cond_signal(&state.qempty_cond);
    pthread_mutex_unlock(&state.qempty_mtx);
 
    alarm(5);
    pthread_join(thread, NULL);
 
    hd_rum_translator_deinit(&state);
    udp_exit(sock_in);
    common_cleanup(init);
    unregister_should_exit_callback(&state.mod, hd_rum_translator_should_exit_callback, const_cast<bool *>(&should_exit));
 
    printf("Exit\n");
 
    return 0;
}

References common_cleanup(), common_preinit(), CONFERENCE, control_init(), control_start(), crash_signal_handler(), create_output_port(), EXIT, EXIT_FAIL_COMPRESS, EXIT_FAIL_CONTROL_SOCK, EXIT_FAIL_DECODER, format_in_si_units(), hang_signal_handler(), hd_rum_decompress_init(), hd_rum_translator_deinit(), hd_rum_translator_should_exit_callback(), init(), invalid_arg_is_numeric(), log_level, LOG_LEVEL_INFO, log_msg(), MSG, parse_fmt(), params::port, print_version(), qinit(), recompress_init(), register_should_exit_callback(), Participant_manager::run_async(), signal_handler(), SIZE, state, Participant_manager::tick(), tv_diff(), udp_exit(), udp_get_udp_rx_port(), udp_init(), udp_init_if(), udp_recvfrom_timeout(), udp_set_recv_buf(), unregister_should_exit_callback(), and writer().

Here is the call graph for this function:

parse_fmt()

static int parse_fmt ( int argc, char ** argv, struct cmdline_parameters * parsed )

static

Todo

Use rather getopt() than manual parsing.

Return values

-1	failure
0	success
1	help shown

Definition at line 728 of file hd-rum-translator.cpp.

{
    const int rc = parse_global_opts(argc, argv, parsed);
    if (rc != 0) {
        return rc;
    }
 
    if (optind + 2 > argc) {
        MSG(FATAL, "Missing parameter%s port!\n\n",
            argc < optind + 1 ? " buffer_size and" : "");
        usage(argv[0]);
        return -1;
    }
 
    parsed->bufsize = parse_size(argv[optind++]);
    try {
        parsed->port = stoi(argv[optind]);
    } catch (invalid_argument &) {
        throw ug_runtime_error(string("invalid port number: ") +
                               argv[optind]);
    }
    if (parsed->port < 0 || parsed->port >= USHRT_MAX) {
        throw ug_runtime_error(string("port must be in range [0..") +
                               to_string(USHRT_MAX) +
                               "], given: " + argv[optind]);
    }
    optind++;
    if (parsed->port == 0) {
        MSG(WARNING,
            "Given RX port 0, reflector will listen on a random port.\n");
    }
 
    parsed->host_count = 0;
    while (optind < argc) {
            parsed->hosts.resize(parsed->host_count + 1);
            parsed->hosts[parsed->host_count].bitrate = RATE_UNLIMITED;
 
            const char *const optstring = "+46P:c:f:l:m:";
            int               ch        = 0;
            while ((ch = getopt(argc, argv, optstring)) != -1) {
                    switch (ch) {
                    case 'P':
                            if (strchr(optarg, ':') != nullptr) {
                                    parsed->hosts[parsed->host_count].rx_port =
                                        stoi(optarg);
                                    parsed->hosts[parsed->host_count].tx_port =
                                        stoi(strchr(optarg, ':') + 1);
                            } else {
                                    parsed->hosts[parsed->host_count].tx_port =
                                        stoi(optarg);
                            }
                            break;
                    case 'm':
                            parsed->hosts[parsed->host_count].common_opts.mtu =
                                stoi(optarg);
                            break;
                    case 'c':
                            parsed->hosts[parsed->host_count].compression = optarg;
                            break;
                    case 'f':
                            parsed->hosts[parsed->host_count].fec = optarg;
                            break;
                    case 'l':
                            if (!parse_bitrate(
                                    optarg, &parsed->hosts[parsed->host_count]
                                                 .bitrate)) {
                                    return -1;
                            }
                            break;
                    case '4':
                            parsed->hosts[parsed->host_count]
                                .common_opts.force_ip_version = 4;
                            break;
                    case '6':
                            parsed->hosts[parsed->host_count]
                                .common_opts.force_ip_version = 6;
                            break;
                    default:
                            MSG(FATAL, "Error: invalid host option.\nn");
                            return -1;
                    }
            }
            if (optind == argc) {
                MSG(ERROR,
                    "Error: host option following last host address!\n");
                return -1;
            }
            parsed->hosts[parsed->host_count++].addr = argv[optind++];
    }
 
    return 0;
}

References MSG, parse_bitrate(), parse_global_opts(), parse_size(), RATE_UNLIMITED, and usage().

Referenced by main().

Here is the call graph for this function:

parse_global_opts()

static int parse_global_opts ( int argc, char ** argv, struct cmdline_parameters * parsed )

static

Definition at line 643 of file hd-rum-translator.cpp.

{
        const struct option getopt_options[] = {
                {"blend",                   no_argument,       nullptr, 'B'},
                { "capture-filter",         required_argument, nullptr, 'F'},
                { "list-modules",           required_argument, nullptr, 'L'},
                { "param",                  required_argument, nullptr, 'O'},
                { "conference-compression", required_argument, nullptr, 'R'},
                { "server",                 required_argument, nullptr, 'S'},
                { "verbose",                optional_argument, nullptr, 'V'},
                { "capabilities",           no_argument,       nullptr, 'b'},
                { "help",                   no_argument,       nullptr, 'h'},
                { "contol-port",            required_argument, nullptr, 'n'},
                { "conference",             required_argument, nullptr, 'r'},
                { "version",                no_argument,       nullptr, 'v'},
                { nullptr,                  0,                 nullptr, 0  }
        };
        const char *const optstring = "+BF:LO:R:S:Vbhn:r:v";
 
        int ch = 0;
        while ((ch = getopt_long(argc, argv, optstring, getopt_options,
                                 nullptr)) != -1) {
                switch (ch) {
                case 'c':
                        parsed->control_port            = stoi(optarg);
                        parsed->control_connection_type = 0;
                        if (strchr(optarg, ':') != nullptr) {
                                parsed->control_connection_type =
                                    stoi(strchr(optarg, ':') + 1);
                        }
                        break;
                case 'b':
                        print_capabilities("");
                        return 1;
                case 'B':
                        parsed->out_conf.mode = BLEND;
                        break;
                case 'r':
                        parsed->out_conf.mode = CONFERENCE;
                        parsed->out_conf.arg  = optarg;
                        break;
                case 'R':
                        parsed->conference_compression = optarg;
                        break;
                case 'S':
                        parsed->server_port = stoi(optarg);
                        break;
                case 'F':
                        parsed->capture_filter = optarg;
                        break;
                case 'h':
                        usage(argv[0]);
                        return 1;
                case 'v':
                        // nothing needed, version is printed everytime
                        return 1;
                case 'V':
                        break; // already handled in common_preinit()
                case 'O':
                        if (!parse_params(optarg, false)) {
                                return 1;
                        }
                        break;
                case 'L':
                        list_all_modules();
                        return 1;
                case '?':
                default:
                        MSG(FATAL, "Unknown global parameter\n\n");
                        usage(argv[0]);
                        return -1;
                }
        }
    return 0;
}

References BLEND, CONFERENCE, list_all_modules(), MSG, parse_params(), print_capabilities(), and usage().

Referenced by parse_fmt().

Here is the call graph for this function:

parse_size()

static int parse_size ( const char * sz_str )

static

unit_evaluate() is similar but uses SI prefixes

Definition at line 614 of file hd-rum-translator.cpp.

{
    int ret = 0;
    size_t end = 0;
 
    try {
        ret = stoi(sz_str, &end);
    } catch (invalid_argument &e) {
        throw ug_runtime_error(string("invalid buffer size: ") + sz_str);
    }
    if (ret <= 0) {
        throw ug_runtime_error(string("size must be positive: ") + sz_str);
    }
 
    switch (sz_str[end]) {
    case 'K':
    case 'k':
        ret *= 1024;
        break;
 
    case 'M':
    case 'm':
        ret *= 1024 * 1024;
        break;
    }
 
    return ret;
}

Referenced by parse_fmt().

qdestroy()

static void qdestroy ( struct item * queue )

static

Definition at line 268 of file hd-rum-translator.cpp.

{
    if (queue == nullptr) {
        return;
    }
 
    struct item *q = queue;
    do {
        free(q->buf);
        q = q->next;
    } while (q != queue);
    free(queue);
}

References item::buf, item::next, and queue.

Referenced by hd_rum_translator_deinit().

qinit()

static struct item * qinit ( int qsize )

static

Definition at line 241 of file hd-rum-translator.cpp.

{
    struct item *queue;
    int i;
 
    if (qsize <= 0) {
        fprintf(stderr, "wrong packet queue size %d items\n", qsize);
        return nullptr;
    }
 
    printf("initializing packet queue for %d items\n", qsize);
 
    queue = (struct item *) calloc(qsize, sizeof(struct item));
    if (queue == NULL) {
        fprintf(stderr, "not enough memory\n");
        exit(2);
    }
 
    for (i = 0; i < qsize; i++) {
        queue[i].buf = (char *) malloc(SIZE);
        queue[i].next = queue + i + 1;
    }
    queue[qsize - 1].next = queue;
 
    return queue;
}

References queue, and SIZE.

Referenced by main().

signal_handler()

static void signal_handler ( int signum )

static

Definition at line 188 of file hd-rum-translator.cpp.

{
#ifdef SIGPIPE
        if (signum == SIGPIPE) {
                signal(SIGPIPE, SIG_IGN);
        }
#endif // defined SIGPIPE
    if (log_level >= LOG_LEVEL_DEBUG) {
        char msg[] = "Caught signal ";
        char buf[128];
        char *ptr = buf;
        for (size_t i = 0; i < sizeof msg - 1; ++i) {
            *ptr++ = msg[i];
        }
        if (signum / 10) {
            *ptr++ = '0' + signum / 10;
        }
        *ptr++ = '0' + signum % 10;
        *ptr++ = '\n';
        size_t bytes = ptr - buf;
        ptr = buf;
        do {
            ssize_t written = write(STDERR_FILENO, ptr, bytes);
            if (written < 0) {
                break;
            }
            bytes -= written;
            ptr += written;
        } while (bytes > 0);
    }
    exit_uv(0);
}

References item::buf, exit_uv(), log_level, and LOG_LEVEL_DEBUG.

Referenced by main().

Here is the call graph for this function:

sockaddr_equal()

static bool sockaddr_equal ( struct sockaddr_storage * a, struct sockaddr_storage * b )

static

Definition at line 839 of file hd-rum-translator.cpp.

                                                                                  {
        if(a->ss_family != b->ss_family)
                return false;
 
        switch(a->ss_family){
        case AF_INET:{
                auto a_in = reinterpret_cast<struct sockaddr_in *>(a);
                auto b_in = reinterpret_cast<struct sockaddr_in *>(b);
                return a_in->sin_port == b_in->sin_port
                        && a_in->sin_addr.s_addr == b_in->sin_addr.s_addr;
        }
        case AF_INET6:{
                auto a_in = reinterpret_cast<struct sockaddr_in6 *>(a);
                auto b_in = reinterpret_cast<struct sockaddr_in6 *>(b);
                return a_in->sin6_port == b_in->sin6_port
                        && memcmp(a_in->sin6_addr.s6_addr, b_in->sin6_addr.s6_addr, sizeof(struct in6_addr)) == 0;
        }
        default:
                assert(false && "Comparison not implemented for address type");
                abort();
        }
}

Referenced by Participant_manager::tick().

usage()

static void usage ( const char * progname )

static

Definition at line 537 of file hd-rum-translator.cpp.

                                        {
    col() << "Usage:\n\t"
          << SBOLD(SRED(progname)
                   << " [global_opts] buffer_size port \\\n\t\t[host1_options] "
                      "host1 [[host2_options] host2] ...")
          << "\n";
 
    col() << "\nwhere:\n"
          << SBOLD("\tbuffer_size")
          << " - network buffer size, eg. 200k for compressed\n\t\t or "
             "8M for uncompressed video\n"
          << SBOLD("\tport") << " - UDP port number\n";
    col() << SUNDERLINE("global_opts") << " may be:\n"
          << SBOLD("\t--control-port|-n <port_number>[:0|:1]")
          << " - control port to connect to, optionally client/server "
             "(default)\n"
          << SBOLD("\t--blend|-B")
          << " - enable blending from original to newly received stream, "
             "increases latency\n"
          << SBOLD("\t--server|-S <port>")
          << " - enable server mode for clients to connect on specified "
             "port\n"
          << SBOLD("\t--conference|-r <width>:<height>[:fps]")
          << " - enable combining of multiple inputs, increases latency\n"
          << SBOLD("\t--conference-compression|-R <compression>")
          << " - compression for conference participants\n"
          << SBOLD("\t--capture-filter|-F <cfg_string>")
          << " - apply video capture filter to incoming video\n"
          << SBOLD("\t--param|-O") << " - additional parameters\n"
          << SBOLD("\t--help|-h\n") << SBOLD("\t--verbose|-V\n") << SBOLD("\t-v")
          << " - print version\n";
    col() << "and " << SUNDERLINE("hostX_options") << " may be:\n"
          << SBOLD("\t-P [<rx_port>:]<tx_port>")
          << " - TX port to be used (optionally also RX)\n"
          << SBOLD("\t-c <compression>") << " - compression\n"
          << "\tFollowing options will be used only if " << SUNDERLINE("'-c'")
          << " parameter is set:\n"
          << SBOLD("\t-m <mtu>") << " - MTU size\n"
          << SBOLD("\t-l <limiting_bitrate>") << " - bitrate to be shaped to\n"
          << SBOLD("\t-f <fec>")
          << " - FEC that will be used for transmission.\n"
          << SBOLD("\t-4/-6") << " - force IPv4/IPv6\n";
    printf("\nPlease note that blending and capture filter is used only "
           "for host for which\n"
           "compression is specified (transcoding is active). If "
           "compression is not\n"
           "set, simple packet retransmission is used. Compression can be "
           "also 'none'\n"
           "for uncompressed transmission (see 'uv -c help' for list).\n");
}

References SBOLD, SRED, and SUNDERLINE.

writer()

static void * writer ( void * arg )

static

Definition at line 366 of file hd-rum-translator.cpp.

{
    struct hd_rum_translator_state *s =
        (struct hd_rum_translator_state *) arg;
 
    while (1) {
        // first check messages
        for (unsigned int i = 0; i < s->replicas.size(); i++) {
            struct message *msg;
            while ((msg = check_message(&s->replicas[i]->mod))) {
                struct response *r = change_replica_type(s, &s->replicas[i]->mod, msg, i);
                free_message(msg, r);
            }
        }
 
        struct msg_universal *msg;
        while ((msg = (struct msg_universal *) check_message(&s->mod))) {
            struct response *r = NULL;
            if (strncasecmp(msg->text, "delete-port ", strlen("delete-port ")) == 0) {
                char *port_spec = msg->text + strlen("delete-port ");
                int index = -1;
                if (isdigit(port_spec[0])) {
                    int i = stoi(port_spec);
                    if (i >= 0 && i < (int) s->replicas.size()) {
                        index = i;
                    } else {
                        log_msg(LOG_LEVEL_WARNING, "Invalid port index: %d. Not removing.\n", i);
                    }
                } else {
                    int i = 0;
                    for (auto r : s->replicas) {
                        if (strcmp(r->mod.name, port_spec) == 0) {
                            index = i;
                            break;
                        }
                        i++;
                    }
                    if (index == -1) {
                        log_msg(LOG_LEVEL_WARNING, "Unknown port name: %s. Not removing.\n", port_spec);
                    }
                }
                if (index >= 0) {
                    recompress_remove_port(s->recompress, index);
                    delete s->replicas[index];
                    s->replicas.erase(s->replicas.begin() + index);
                    log_msg(LOG_LEVEL_NOTICE, "Deleted output port %d.\n", index);
                }
            } else if (strncasecmp(msg->text, "create-port", strlen("create-port")) == 0) {
                // format of parameters is either:
                // <host>:<port> [<compression>]
                // or (for compat with older CoUniverse version)
                // <host> <port> [<compression>]
                char *host_port, *port_str = NULL, *save_ptr;
                char *host;
                int tx_port;
                strtok_r(msg->text, " ", &save_ptr);
                host_port = strtok_r(NULL, " ", &save_ptr);
                if (host_port && (strchr(host_port, ':') != NULL || (port_str = strtok_r(NULL, " ", &save_ptr)) != NULL)) {
                    if (port_str) {
                        host = host_port;
                        tx_port = stoi(port_str);
                    } else {
                        tx_port = stoi(strrchr(host_port, ':') + 1);
                        host = host_port;
                        *strrchr(host_port, ':') = '\0';
                    }
                    // handle square brackets around an IPv6 address
                    if (host[0] == '[' && host[strlen(host) - 1] == ']') {
                        host += 1;
                        host[strlen(host) - 1] = '\0';
                    }
                } else {
                    const char *err_msg = "wrong format";
                    log_msg(LOG_LEVEL_ERROR, "%s\n", err_msg);
                    free_message((struct message *) msg, new_response(RESPONSE_BAD_REQUEST, err_msg));
                    continue;
                }
                char *compress = strtok_r(NULL, " ", &save_ptr);
 
                struct common_opts opts = { COMMON_OPTS_INIT };
                int idx = create_output_port(s,
                        host, 0, tx_port, s->bufsize, &opts,
                        compress, nullptr, RATE_UNLIMITED, s->server_socket != nullptr);
 
                if(idx < 0) {
                    free_message((struct message *) msg, new_response(RESPONSE_INT_SERV_ERR, "Cannot create output port."));
                    continue;
                }
 
                if(compress)
                    log_msg(LOG_LEVEL_NOTICE, "Created new transcoding output port %s:%d:0x%08" PRIx32 ".\n", host, tx_port, recompress_get_port_ssrc(s->recompress, idx));
                else
                    log_msg(LOG_LEVEL_NOTICE, "Created new forwarding output port %s:%d.\n", host, tx_port);
 
            } else {
                r = new_response(RESPONSE_BAD_REQUEST, NULL);
            }
 
            free_message((struct message *) msg, r ? r : new_response(RESPONSE_OK, NULL));
        }
 
        // then process incoming packets
        while (s->qhead != s->qtail) {
            if(s->qhead->size == 0) { // poisoned pill
                return NULL;
            }
 
            // pass it for transcoding if needed
            if (recompress_get_num_active_ports(s->recompress) > 0) {
                ssize_t ret = hd_rum_decompress_write(s->decompress, s->qhead->buf, s->qhead->size);
                if (ret < 0) {
                    perror("hd_rum_decompress_write");
                }
            }
 
            // distribute it to output ports that don't need transcoding
#ifdef _WIN32
            // send it asynchronously in MSW (performance optimalization)
            SleepEx(0, TRUE); // allow system to call our completion routines in APC
            int ref = 0;
            for (unsigned int i = 0; i < s->replicas.size(); i++) {
                if(s->replicas[i]->type == replica::type_t::USE_SOCK) {
                    ref++;
                }
            }
            struct wsa_aux_storage *aux = (struct wsa_aux_storage *)(void *) ((char *) s->qhead->buf + OFFSET);
            memset(aux, 0, sizeof *aux);
            aux->overlapped = (WSAOVERLAPPED *) calloc(ref, sizeof(WSAOVERLAPPED));
            aux->ref = ref;
            int overlapped_idx = 0;
            for (unsigned int i = 0; i < s->replicas.size(); i++) {
                if(s->replicas[i]->type == replica::type_t::USE_SOCK) {
                    aux->overlapped[overlapped_idx].hEvent = s->qhead->buf;
                    ssize_t ret = udp_sendto_wsa_async(s->replicas[i]->sock.get(), s->qhead->buf, s->qhead->size,
                                    wsa_deleter, &aux->overlapped[overlapped_idx], (sockaddr *) &s->replicas[i]->sockaddr, s->replicas[i]->sockaddr_len);
                    if (ret < 0) {
                        perror("Hd-rum-translator send");
                    }
                    overlapped_idx += 1;
                }
            }
            // reallocate the buffer since the last one will be freeed automaticaly
            s->qhead->buf = (char *) malloc(SIZE);
#else
            for (unsigned int i = 0; i < s->replicas.size(); i++) {
                if(s->replicas[i]->type == replica::type_t::USE_SOCK) {
                    ssize_t ret = udp_sendto(s->replicas[i]->sock.get(), s->qhead->buf, s->qhead->size, (sockaddr *) &s->replicas[i]->sockaddr, s->replicas[i]->sockaddr_len);
                    if (ret < 0) {
                        perror("Hd-rum-translator send");
                    }
                }
            }
#endif
            s->qhead = s->qhead->next;
 
            pthread_mutex_lock(&s->qfull_mtx);
            s->qfull = 0;
            pthread_cond_signal(&s->qfull_cond);
            pthread_mutex_unlock(&s->qfull_mtx);
        }
 
        pthread_mutex_lock(&s->qempty_mtx);
        if (s->qempty)
            pthread_cond_wait(&s->qempty_cond, &s->qempty_mtx);
        s->qempty = 1;
        pthread_mutex_unlock(&s->qempty_mtx);
    }
 
    return NULL;
}

References item::buf, hd_rum_translator_state::bufsize, change_replica_type(), check_message(), COMMON_OPTS_INIT, create_output_port(), hd_rum_translator_state::decompress, free_message(), hd_rum_decompress_write(), LOG_LEVEL_ERROR, LOG_LEVEL_NOTICE, LOG_LEVEL_WARNING, log_msg(), hd_rum_translator_state::mod, msg, new_response(), item::next, hd_rum_translator_state::qempty, hd_rum_translator_state::qempty_cond, hd_rum_translator_state::qempty_mtx, hd_rum_translator_state::qfull, hd_rum_translator_state::qfull_cond, hd_rum_translator_state::qfull_mtx, hd_rum_translator_state::qhead, hd_rum_translator_state::qtail, RATE_UNLIMITED, hd_rum_translator_state::recompress, recompress_get_num_active_ports(), recompress_get_port_ssrc(), recompress_remove_port(), hd_rum_translator_state::replicas, RESPONSE_BAD_REQUEST, RESPONSE_INT_SERV_ERR, RESPONSE_OK, hd_rum_translator_state::server_socket, SIZE, item::size, TRUE, and udp_sendto().

Referenced by main().

Here is the call graph for this function: