video_compress.cpp

Go to the documentation of this file.

 
/*
 * Copyright (c) 2011-2024 CESNET z.s.p.o.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, is permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * 3. Neither the name of CESNET nor the names of its contributors may be
 *    used to endorse or promote products derived from this software without
 *    specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING,
 * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
 * EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
 * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
 
#include <cassert>
#include <cinttypes>
#include <cstdio>
#include <cstring>
#include <memory>
#include <string>
#include <thread>
#include <vector>
 
#include "messaging.h"
#include "module.h"
#include "tv.h"
#include "utils/synchronized_queue.h"
#include "utils/thread.h"
#include "utils/vf_split.h"
#include "utils/worker.h"
#include "video.h"
#include "video_compress.h"
#include "lib_common.h"
#include "debug.h"
 
#define MOD_NAME "[vcompress] "
 
using namespace std;
 
struct compress_state;
 
namespace {
struct compress_state_real {
private:
        compress_state_real(struct module *parent, const char *config_string);
        void          start(struct compress_state *proxy);
        void          async_consumer(struct compress_state *s);
        void          async_tile_consumer(struct compress_state *s);
        thread        asynch_consumer_thread;
public:
        static compress_state_real *create(struct module *parent, const char *config_string,
                        struct compress_state *proxy) {
                compress_state_real *s = new compress_state_real(parent, config_string);
                s->start(proxy);
                return s;
        }
        ~compress_state_real();
        const video_compress_info    *funcs;            
        vector<struct module *> state;                  
        string              compress_options; 
        volatile bool       discard_frames;   
};
}
 
struct compress_state {
        struct module mod;               
        struct compress_state_real *ptr; 
        synchronized_queue<shared_ptr<video_frame>, 1> queue;
        bool poisoned = false;
};
 
static shared_ptr<video_frame> compress_frame_tiles(struct compress_state *proxy,
                shared_ptr<video_frame> frame);
 
void show_compress_help(bool full)
{
        printf("Possible compression modules (see '-c <module>:help' for options):\n");
        list_modules(LIBRARY_CLASS_VIDEO_COMPRESS, VIDEO_COMPRESS_ABI_VERSION, full);
}
 
static void async_poison(struct compress_state_real *s){
        if (s->funcs->compress_frame_async_push_func) {
                s->funcs->compress_frame_async_push_func(s->state[0], {}); // poison
        } else if (s->funcs->compress_tile_async_push_func){
                for(size_t i = 0; i < s->state.size(); i++){
                        s->funcs->compress_tile_async_push_func(s->state[i], {}); // poison
                }
        }
}
 
static void compress_process_message(struct compress_state *proxy, struct msg_change_compress_data *data)
{
        struct response *r = NULL;
        /* In this case we are only changing some parameter of compression.
         * This means that we pass the parameter to compress driver. */
        if(data->what == CHANGE_PARAMS) {
                for(unsigned int i = 0; i < proxy->ptr->state.size(); ++i) {
                        struct msg_change_compress_data *tmp_data =
                                (struct msg_change_compress_data *)
                                new_message(sizeof(struct msg_change_compress_data));
                        tmp_data->what = data->what;
                        strncpy(tmp_data->config_string, data->config_string,
                                        sizeof(tmp_data->config_string));
                        struct response *resp = send_message_to_receiver(proxy->ptr->state[i],
                                        (struct message *) tmp_data);
                        free_response(r); // frees previous response
                        r = resp;
                }
 
        } else {
                struct compress_state_real *new_state;
                char config[1024];
                strncpy(config, data->config_string, sizeof(config));
 
                try {
                        new_state = compress_state_real::create(&proxy->mod, config, proxy);
                } catch (int i) {
                        free_message((struct message *) data,
                                        new_response(RESPONSE_INT_SERV_ERR, NULL));
                        return;
                }
 
                struct compress_state_real *old = proxy->ptr;
                // let the async processing finish
                old->discard_frames = true;
                async_poison(old);
                delete old;
                proxy->ptr = new_state;
                r = new_response(RESPONSE_OK, NULL);
        }
 
        free_message((struct message *) data, r);
}
 
int compress_init(struct module *parent, const char *config_string, struct compress_state **state) {
        struct compress_state *proxy;
        proxy = new struct compress_state();
 
        module_init_default(&proxy->mod);
        proxy->mod.cls = MODULE_CLASS_COMPRESS;
        proxy->mod.priv_data = proxy;
 
        try {
                proxy->ptr = compress_state_real::create(&proxy->mod, config_string, proxy);
        } catch (int i) {
                delete proxy;
                return i;
        }
 
        module_register(&proxy->mod, parent);
 
        *state = proxy;
        return 0;
}
 
compress_state_real::compress_state_real(struct module *parent, const char *config_string) :
        funcs(nullptr), discard_frames(false)
{
        string compress_name;
 
        if (!config_string)
                throw -1;
 
        if (strcmp(config_string, "help") == 0 || strcmp(config_string, "fullhelp") == 0) {
                show_compress_help(strcmp(config_string, "fullhelp") == 0);
                throw 1;
        }
 
        char *tmp = strdup(config_string);
        if (strchr(tmp, ':')) {
                char *opts = strchr(tmp, ':') + 1;
                *strchr(tmp, ':') = '\0';
                compress_options = opts;
        }
        compress_name = tmp;
        free(tmp);
 
        auto vci = static_cast<const struct video_compress_info *>(load_library(compress_name.c_str(), LIBRARY_CLASS_VIDEO_COMPRESS, VIDEO_COMPRESS_ABI_VERSION));
        if(!vci) {
                LOG(LOG_LEVEL_ERROR)
                    << MOD_NAME "Unknown or unavailable compression: "
                    << config_string << "\n";
                throw -1;
        }
 
        funcs = vci;
 
        if (funcs->init_func) {
                state.resize(1);
                state[0] = funcs->init_func(parent, compress_options.c_str());
                if(!state[0]) {
                        LOG(LOG_LEVEL_ERROR)
                            << MOD_NAME "Compression initialization failed: "
                            << config_string << "\n";
                        throw -1;
                }
                if(state[0] == INIT_NOERR) {
                        throw 1;
                }
        } else {
                throw -1;
        }
}
 
void compress_state_real::start(struct compress_state *proxy)
{
        if (funcs->compress_frame_async_push_func) {
                asynch_consumer_thread = thread(&compress_state_real::async_consumer, this, proxy);
        } else if (funcs->compress_tile_async_push_func){
                asynch_consumer_thread = thread(&compress_state_real::async_tile_consumer, this, proxy);
        }
}
 
static bool check_state_count(unsigned tile_count, struct compress_state *proxy)
{
        struct compress_state_real *s = proxy->ptr;
 
        if(tile_count != s->state.size()) {
                size_t old_size = s->state.size();
                s->state.resize(tile_count);
                for (unsigned int i = old_size; i < s->state.size(); ++i) {
                        s->state[i] = s->funcs->init_func(&proxy->mod, s->compress_options.c_str());
                        if(!s->state[i]) {
                                LOG(LOG_LEVEL_ERROR) << MOD_NAME
                                    "Compression initialization failed\n";
                                return false;
                        }
                }
        }
        return true;
}
 
void compress_frame(struct compress_state *proxy, shared_ptr<video_frame> frame)
{
        if (!proxy)
                abort();
 
        struct msg_change_compress_data *msg = NULL;
        while ((msg = (struct msg_change_compress_data *) check_message(&proxy->mod))) {
                compress_process_message(proxy, msg);
        }
 
        struct compress_state_real *s = proxy->ptr;
 
        if (!frame) {
                proxy->poisoned = true;
        }
        if (frame) {
                frame->compress_start = get_time_in_ns();
        }
 
        if (s->funcs->compress_frame_async_push_func) {
                assert(s->funcs->compress_frame_async_pop_func);
                s->funcs->compress_frame_async_push_func(s->state[0], frame);
                return;
        }
        if (s->funcs->compress_tile_async_push_func) {
                assert(s->funcs->compress_tile_async_pop_func);
                if (!frame) {
                        async_poison(s);
                        return;
                }
 
                if(!check_state_count(frame->tile_count, proxy)){
                        return;
                }
 
                vector<shared_ptr<video_frame>> separate_tiles = vf_separate_tiles(frame);
 
                for(unsigned i = 0; i < separate_tiles.size(); i++){
                        s->funcs->compress_tile_async_push_func(s->state[i], separate_tiles[i]);
                }
                return;
        }
 
        // sync APIs - pass poisoned pill to the queue but not to compressions,
        if (!frame) { // which doesn't need that but use NULL frame differently
                proxy->queue.push(shared_ptr<video_frame>());
                return;
        }
 
        shared_ptr<video_frame> sync_api_frame;
        do {
                if (s->funcs->compress_frame_func) {
                        sync_api_frame = s->funcs->compress_frame_func(s->state[0], frame);
                } else if(s->funcs->compress_tile_func) {
                        sync_api_frame = compress_frame_tiles(proxy, frame);
                } else {
                        assert(!"No eligible compress API found");
                }
 
                // empty return value here represents error, but we don't want to pass it to queue, since it would
                // be interpreted as poisoned pill
                if (!sync_api_frame) {
                        return;
                }
                sync_api_frame->compress_end = get_time_in_ns();
                proxy->queue.push(sync_api_frame);
                frame = nullptr;
        } while (true);
}
 
struct compress_worker_data {
        struct module *state;      
        shared_ptr<video_frame> frame; 
 
        compress_tile_t callback;  
        shared_ptr<video_frame> ret; 
};
 
static void *compress_tile_callback(void *arg) {
        compress_worker_data *s = (compress_worker_data *) arg;
 
        s->ret = s->callback(s->state, s->frame);
 
        return s;
}
 
static shared_ptr<video_frame> compress_frame_tiles(struct compress_state *proxy,
                shared_ptr<video_frame> frame)
{
        struct compress_state_real *s = proxy->ptr;
        const int tile_cnt = (int) proxy->ptr->state.size();
        vector<shared_ptr<video_frame>> separate_tiles;
        if (frame) {
                if (!check_state_count(frame->tile_count, proxy)) {
                        return nullptr;
                }
                separate_tiles = vf_separate_tiles(frame);
        } else {
                separate_tiles.resize(tile_cnt);
        }
 
        // frame pointer may no longer be valid
        frame = NULL;
 
        vector<task_result_handle_t> task_handle(tile_cnt);
 
        vector <compress_worker_data> data_tile(tile_cnt);
        for (int i = 0; i < tile_cnt; ++i) {
                struct compress_worker_data *data = &data_tile[i];
                data->state = s->state[i];
                data->frame = separate_tiles[i];
                data->callback = s->funcs->compress_tile_func;
 
                task_handle[i] = task_run_async(compress_tile_callback, data);
        }
 
        vector<shared_ptr<video_frame>> compressed_tiles(separate_tiles.size());
 
        bool failed = false;
        for(unsigned int i = 0; i < separate_tiles.size(); ++i) {
                struct compress_worker_data *data = (struct compress_worker_data *)
                        wait_task(task_handle[i]);
 
                if(!data->ret) {
                        failed = true;
                }
 
                compressed_tiles[i] = data->ret;
        }
 
        if (failed) {
                return NULL;
        }
 
        return vf_merge_tiles(compressed_tiles);
}
void
compress_done(struct compress_state *proxy)
{
        if (proxy == nullptr) {
                return;
        }
 
        struct compress_state_real *s = proxy->ptr;
        if (!proxy->poisoned) { // pass poisoned pill if it wasn't
                compress_frame(proxy, {});
        }
 
        delete s;
        module_done(&proxy->mod);
        delete proxy;
}
 
compress_state_real::~compress_state_real()
{
        if (asynch_consumer_thread.joinable()) {
                asynch_consumer_thread.join();
        }
 
        for(unsigned int i = 0; i < state.size(); ++i) {
                module_done(state[i]);
        }
}
 
namespace {
void compress_state_real::async_tile_consumer(struct compress_state *s)
{
        set_thread_name(__func__);
        vector<shared_ptr<video_frame>> compressed_tiles;
        unsigned expected_seq = 0;
        while (true) {
                bool fail = false;
                for(unsigned i = 0; i < state.size(); i++){
                        std::shared_ptr<video_frame> ret = nullptr;
                        //discard frames with seq lower than expected
                        do {
                                ret = funcs->compress_tile_async_pop_func(state[i]);
                        } while(ret && ret->seq < expected_seq);
 
                        if (!ret) {
                                if(!discard_frames)
                                        s->queue.push(nullptr); //poison
                                return;
                        }
 
                        if(ret->seq > expected_seq){
                                log_msg(LOG_LEVEL_ERROR,
                                                "Expected sequence number %u but got %u!\n",
                                                expected_seq,
                                                ret->seq);
                                if(i == 0){
                                        //If this is first tile we can continue
                                        expected_seq = ret->seq;
                                } else {
                                        expected_seq = ret->seq + 1;
                                        fail = true;
                                        break;
                                }
                        }
 
                        ret->compress_end = get_time_in_ns();
                        compressed_tiles.resize(state.size(), nullptr);
                        compressed_tiles[i] = std::move(ret);
                }
 
                if(fail)
                        continue;
 
                if (!discard_frames) {
                        s->queue.push(vf_merge_tiles(compressed_tiles));
                }
                //If frames are not numbered they always have seq = 0
                if(expected_seq > 0) expected_seq++;
        }
}
 
void compress_state_real::async_consumer(struct compress_state *s)
{
        set_thread_name(__func__);
        while (true) {
                auto frame = funcs->compress_frame_async_pop_func(state[0]);
                if (!discard_frames) {
                        s->queue.push(frame);
 
                }
                if (!frame) {
                        return;
 
                }
 
        }
 
}
} // end of anonymous namespace
 
shared_ptr<video_frame> compress_pop(struct compress_state *proxy)
{
        if(!proxy)
                return NULL;
 
        auto f = proxy->queue.pop();
        if (f) {
                MSG(DEBUG, "Compressed frame size: %8u; duration: %7.3f ms\n",
                    vf_get_data_len(f.get()),
                    (f->compress_end - f->compress_start) / MS_IN_NS_DBL);
        }
        return f;
}