hd-rum-recompress.cpp File Reference

#include <cinttypes>
#include <chrono>
#include <memory>
#include <thread>
#include <string>
#include "hd-rum-translator/hd-rum-recompress.h"
#include "debug.h"
#include "host.h"
#include "utils/misc.h"
#include "rtp/rtp.h"
#include "tv.h"
#include "video_compress.h"
#include "video_rxtx/ultragrid_rtp.hpp"
#include "utils/profile_timer.hpp"

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

Go to the source code of this file.

Classes
struct	recompress_output_port
struct	recompress_worker_ctx
struct	state_recompress

Functions
static void	extract_port (struct state_recompress *s, const std::string &compress_cfg, int i, recompress_output_port *move_to=nullptr)
static int	move_port_to_worker (struct state_recompress *s, const char *compress, recompress_output_port &&port)
int	recompress_add_port (struct state_recompress *s, const char *host, const char *compress, unsigned short rx_port, unsigned short tx_port, const struct common_opts *common, const char *fec, long long bitrate)
void	recompress_done (struct state_recompress *s)
int	recompress_get_num_active_ports (struct state_recompress *s)
uint32_t	recompress_get_port_ssrc (struct state_recompress *s, int idx)
struct state_recompress *	recompress_init (struct module *parent)
bool	recompress_port_change_compress (struct state_recompress *s, int index, const char *new_compress)
void	recompress_port_set_active (struct state_recompress *s, int index, bool active)
static void	recompress_port_write (recompress_output_port &port, shared_ptr< video_frame > frame)
void	recompress_process_async (state_recompress *s, std::shared_ptr< video_frame > frame)
void	recompress_remove_port (struct state_recompress *s, int index)
static void	recompress_worker (struct recompress_worker_ctx *ctx)
static int	worker_get_num_active_ports (const recompress_worker_ctx &worker)

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

Component of the transcoding reflector that takes an uncompressed frame, recompresses it to another compression and sends it to destination (therefore it wraps the whole sending part of UltraGrid).

Definition in file hd-rum-recompress.cpp.

Function Documentation

extract_port()

static void extract_port ( struct state_recompress * s, const std::string & compress_cfg, int i, recompress_output_port * move_to = nullptr )

static

Definition at line 230 of file hd-rum-recompress.cpp.

{
        auto& worker = s->workers[compress_cfg];
        {
                std::unique_lock<std::mutex> lock(worker.ports_mut);
                if(move_to)
                        *move_to = std::move(worker.ports[i]);
                worker.ports.erase(worker.ports.begin() + i);
 
                if(worker.ports.empty()){
                        //poison compress
                        compress_frame(worker.compress.get(), nullptr);
                        worker.thread.join();
                        s->workers.erase(compress_cfg);
                }
        }
 
        for(auto& p : s->index_to_port){
                if(p.first == compress_cfg && p.second > i)
                        p.second--;
        }
}

References compress_frame(), state_recompress::index_to_port, module::lock, worker(), and state_recompress::workers.

Referenced by recompress_port_change_compress(), and recompress_remove_port().

Here is the call graph for this function:

move_port_to_worker()

static int move_port_to_worker ( struct state_recompress * s, const char * compress, recompress_output_port && port )

static

Definition at line 183 of file hd-rum-recompress.cpp.

{
        auto& worker = s->workers[compress];
        if(!worker.compress){
                worker.compress_cfg = compress;
                int ret = compress_init(s->parent, compress, out_ptr(worker.compress));
                if (ret != 0) {
                        s->workers.erase(compress);
                        return -1;
                }
 
                worker.thread = std::thread(recompress_worker, &worker);
        }
 
        std::lock_guard<std::mutex> lock(worker.ports_mut);
        int index_in_worker = worker.ports.size();
        worker.ports.push_back(std::move(port));
 
        return index_in_worker;
}

References compress_init(), module::lock, state_recompress::parent, recompress_worker(), worker(), and state_recompress::workers.

Referenced by recompress_add_port(), and recompress_port_change_compress().

Here is the call graph for this function:

recompress_add_port()

int recompress_add_port	(	struct state_recompress *	s,
		const char *	host,
		const char *	compress,
		unsigned short	rx_port,
		unsigned short	tx_port,
		const struct common_opts *	common,
		const char *	fec,
		long long	bitrate )

Definition at line 205 of file hd-rum-recompress.cpp.

{
        recompress_output_port port;
 
        try{
                port = recompress_output_port(host, rx_port, tx_port,
                                common, fec, bitrate);
        } catch(...) {
                return -1;
        }
 
        std::lock_guard<std::mutex> lock(s->mut);
        int index_in_worker = move_port_to_worker(s, compress, std::move(port));
        if(index_in_worker < 0)
                return -1;
 
        int index_of_port = s->index_to_port.size();
        s->index_to_port.emplace_back(compress, index_in_worker);
 
        return index_of_port;
}

References state_recompress::index_to_port, module::lock, move_port_to_worker(), and state_recompress::mut.

Referenced by create_output_port().

Here is the call graph for this function:

recompress_done()

void recompress_done

(

struct state_recompress *

s

)

Definition at line 342 of file hd-rum-recompress.cpp.

                                                 {
        {
                std::lock_guard<std::mutex> lock(s->mut);
                for(auto& worker : s->workers){
                        //poison compress
                        compress_frame(worker.second.compress.get(), nullptr);
 
                        worker.second.thread.join();
                        compress_done(worker.second.compress.get());
                }
        }
        delete s;
}

References compress_done(), compress_frame(), state_recompress::mut, worker(), and state_recompress::workers.

Referenced by hd_rum_translator_deinit().

Here is the call graph for this function:

recompress_get_num_active_ports()

int recompress_get_num_active_ports

(

struct state_recompress *

s

)

Definition at line 313 of file hd-rum-recompress.cpp.

                                                               {
        std::lock_guard<std::mutex> lock(s->mut);
        int ret = 0;
        for(const auto& worker : s->workers){
                ret += worker_get_num_active_ports(worker.second);
        }
 
        return ret;
}

References module::lock, state_recompress::mut, worker(), worker_get_num_active_ports(), and state_recompress::workers.

Referenced by writer().

Here is the call graph for this function:

recompress_get_port_ssrc()

uint32_t recompress_get_port_ssrc	(	struct state_recompress *	s,
		int	idx )

Definition at line 263 of file hd-rum-recompress.cpp.

                                                                      {
        std::lock_guard<std::mutex> lock(s->mut);
        auto [compress_cfg, i] = s->index_to_port[idx];
 
        std::lock_guard<std::mutex> work_lock(s->workers[compress_cfg].ports_mut);
        return s->workers[compress_cfg].ports[i].video_rxtx->get_ssrc();
}

References state_recompress::index_to_port, module::lock, state_recompress::mut, and state_recompress::workers.

Referenced by writer().

recompress_init()

struct state_recompress * recompress_init

(

struct module *

parent

)

Definition at line 323 of file hd-rum-recompress.cpp.

                                                                {
        auto state = new state_recompress();
        if(!state)
                return nullptr;
 
        state->parent = parent;
 
        return state;
}

References parent, and state.

Referenced by main().

recompress_port_change_compress()

bool recompress_port_change_compress	(	struct state_recompress *	s,
		int	index,
		const char *	new_compress )

Definition at line 281 of file hd-rum-recompress.cpp.

{
        std::lock_guard<std::mutex> lock(s->mut);
        auto [old_compress, i] = s->index_to_port[index];
 
        if(old_compress == new_compress)
                return true;
 
        recompress_output_port port;
        extract_port(s, old_compress, i, &port);
        int index_in_worker = move_port_to_worker(s, new_compress, std::move(port));
 
        if(index_in_worker < 0){
                s->index_to_port.erase(s->index_to_port.begin() + index);
                return false;
        }
 
        s->index_to_port[index] = {new_compress, index_in_worker};
 
        return true;
}

References extract_port(), state_recompress::index_to_port, module::lock, move_port_to_worker(), and state_recompress::mut.

Referenced by change_replica_type().

Here is the call graph for this function:

recompress_port_set_active()

void recompress_port_set_active	(	struct state_recompress *	s,
		int	index,
		bool	active )

Definition at line 271 of file hd-rum-recompress.cpp.

{
        std::lock_guard<std::mutex> lock(s->mut);
        auto [compress_cfg, i] = s->index_to_port[index];
 
        std::unique_lock<std::mutex> worker_lock(s->workers[compress_cfg].ports_mut);
        s->workers[compress_cfg].ports[i].active = active;
}

References state_recompress::index_to_port, module::lock, state_recompress::mut, and state_recompress::workers.

Referenced by change_replica_type(), and create_output_port().

recompress_port_write()

static void recompress_port_write ( recompress_output_port & port, shared_ptr< video_frame > frame )

static

Definition at line 146 of file hd-rum-recompress.cpp.

{
        PROFILE_FUNC;
 
        port.frames += 1;
 
        auto now = chrono::steady_clock::now();
 
        double seconds = chrono::duration_cast<chrono::duration<double>>(now - port.t0).count();
        if(seconds > 5) {
                double fps = port.frames / seconds;
                log_msg(LOG_LEVEL_INFO, "[0x%08" PRIx32 "->%s:%d:0x%08" PRIx32 "] %d frames in %g seconds = %g FPS\n",
                                frame->ssrc,
                                port.host.c_str(), port.tx_port,
                                port.video_rxtx->get_ssrc(),
                                port.frames, seconds, fps);
                port.t0 = now;
                port.frames = 0;
        }
 
        port.video_rxtx->send(std::move(frame));
}

References count, frame, recompress_output_port::frames, recompress_output_port::host, LOG_LEVEL_INFO, log_msg(), PROFILE_FUNC, video_frame::ssrc, recompress_output_port::t0, recompress_output_port::tx_port, and recompress_output_port::video_rxtx.

Referenced by recompress_worker().

Here is the call graph for this function:

recompress_process_async()

void recompress_process_async	(	state_recompress *	s,
		std::shared_ptr< video_frame >	frame )

Definition at line 333 of file hd-rum-recompress.cpp.

                                                                                    {
        PROFILE_FUNC;
        std::lock_guard<std::mutex> lock(s->mut);
        for(const auto& worker : s->workers){
                if(worker_get_num_active_ports(worker.second) > 0)
                        compress_frame(worker.second.compress.get(), frame);
        }
}

References compress_frame(), frame, state_recompress::mut, PROFILE_FUNC, worker(), worker_get_num_active_ports(), and state_recompress::workers.

Referenced by hd_rum_decompress::state_transcoder_decompress::worker().

Here is the call graph for this function:

recompress_remove_port()

void recompress_remove_port	(	struct state_recompress *	s,
		int	index )

Definition at line 255 of file hd-rum-recompress.cpp.

                                                                  {
        std::lock_guard<std::mutex> lock(s->mut);
        auto [compress_cfg, i] = s->index_to_port[index];
 
        extract_port(s, compress_cfg, i);
        s->index_to_port.erase(s->index_to_port.begin() + index);
}

References extract_port(), state_recompress::index_to_port, module::lock, and state_recompress::mut.

Referenced by writer().

Here is the call graph for this function:

recompress_worker()

static void recompress_worker ( struct recompress_worker_ctx * ctx )

static

Definition at line 169 of file hd-rum-recompress.cpp.

                                                                {
        PROFILE_FUNC;
        assert(ctx->compress);
 
        while(auto frame = compress_pop(ctx->compress.get())){
                std::lock_guard<std::mutex> lock(ctx->ports_mut);
                for(auto& port : ctx->ports){
                        if(port.active)
                                recompress_port_write(port, frame);
                }
                PROFILE_DETAIL("compress_pop");
        }
}

References recompress_worker_ctx::compress, compress_pop(), frame, module::lock, recompress_worker_ctx::ports, recompress_worker_ctx::ports_mut, PROFILE_DETAIL, PROFILE_FUNC, and recompress_port_write().

Referenced by move_port_to_worker().

Here is the call graph for this function:

worker_get_num_active_ports()

static int worker_get_num_active_ports ( const recompress_worker_ctx & worker )

static

Definition at line 304 of file hd-rum-recompress.cpp.

                                                                           {
        int ret = 0;
        for(const auto& port : worker.ports){
                if(port.active)
                        ret++;
        }
        return ret;
}

References worker().

Referenced by recompress_get_num_active_ports(), and recompress_process_async().

Here is the call graph for this function: