src/gstamldmxwrap.c

/* GStreamer
 * Copyright (C) 2023 <xuesong.jiang@amlogic.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include "gstamldmxwrap.h"
#include "gstamldmabufmgrwrap.h"

#include <stdio.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <sys/prctl.h>

#include <poll.h>
#include <fcntl.h>
#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define DMX_COUNT (3)
#define DMX_FILTER_COUNT (32 * DMX_COUNT)
#define SEC_BUF_SIZE (4096)
#define NON_SEC_BUF_HEADER_SIZE (32)
#define SEC_BUF_HEADER_SIZE (40)
#define DMX_POLL_TIMEOUT (200)

typedef struct
{
    gint dev_no;
    gint fd;
    gint used;
    gint enable;
    gint need_free;
    guint flags;
    gst_amldmxwrap_data_cb cb;
    void *user_data;
} dvb_dmx_filter_t;

typedef struct
{
    gint dmx_fd;
    gint dmx_src_fd;
    gint dev_no;
    gint running;
    GstTask *task;
    GRecMutex tlock;
    GMutex lock;

    dvb_dmx_filter_t filter[DMX_FILTER_COUNT];
} dvb_dmx_t;

static dvb_dmx_t dmx_devices[DMX_COUNT];

static int amsysfs_set_sysfs_str(const char *path, const char *val)
{
    int fd;
    fd = open(path, O_CREAT | O_RDWR | O_TRUNC, 0644);
    if (fd >= 0)
    {
        write(fd, val, strlen(val));
        close(fd);
        return 0;
    }
    return -1;
}

static int set_dmx_source()
{
    amsysfs_set_sysfs_str("/sys/class/stb/source", "dmx0");
    amsysfs_set_sysfs_str("/sys/class/stb/demux0_source", "hiu");
    return 0;
}

static inline DVB_RESULT dmx_get_dev(gint dev_no, dvb_dmx_t **dev)
{
    if ((dev_no < 0) || (dev_no >= DMX_COUNT))
    {
        GST_DEBUG("invalid demux device number %d, must in(%d~%d)", dev_no, 0, DMX_COUNT - 1);
        return DVB_FAILURE;
    }

    *dev = &dmx_devices[dev_no];
    return DVB_SUCCESS;
}

static void *gst_amldmxwrap_data_thread(void *arg)
{
    gint i, fid;
    gint ret;
    gint cnt, len;
    guint mask;
    gchar *sec_buf = NULL;
    gint fids[DMX_FILTER_COUNT];
    struct pollfd fds[DMX_FILTER_COUNT];
    dvb_dmx_filter_t *filter = NULL;
    dvb_dmx_t *dmx = (dvb_dmx_t *)arg;

    sec_buf = (gchar *)malloc(SEC_BUF_SIZE);
    prctl(PR_SET_NAME, "gst_amldmxwrap_data_thread");
    while (g_atomic_int_get(&dmx->running))
    {
        cnt = 0;
        mask = 0;

        g_mutex_lock(&dmx->lock);
        for (fid = 0; fid < DMX_FILTER_COUNT; fid++)
        {
            if (dmx->filter[fid].need_free)
            {
                filter = &dmx->filter[fid];
                close(filter->fd);
                filter->used = 0;
                filter->need_free = 0;
                filter->cb = NULL;
            }

            if (dmx->filter[fid].used)
            {
                fds[cnt].events = POLLIN | POLLERR;
                // It is not necessary to protect code block below with
                // Record_DeviceContext_t.lock, so the following annotation
                // is given to suppress related Coverity complaint.
                // coverity[missing_lock]
                fds[cnt].fd = dmx->filter[fid].fd;
                fids[cnt] = fid;
                cnt++;
            }
        }

        g_mutex_unlock(&dmx->lock);

        if (!cnt)
        {
            // g_usleep(20 * 1000);
            g_usleep(1000);
            continue;
        }

        ret = poll(fds, cnt, DMX_POLL_TIMEOUT);
        if (ret <= 0)
        {
            continue;
        }

        for (i = 0; i < cnt; i++)
        {
            if (fds[i].revents & (POLLIN | POLLERR))
            {
                g_mutex_lock(&dmx->lock);
                filter = &dmx->filter[fids[i]];
                if (!filter->enable || !filter->used || filter->need_free)
                {
                    GST_DEBUG("ch[%d] not used, not read", fids[i]);
                    len = 0;
                }
                else
                {

                    gint read_len = 0;

                    if ((filter->flags & DMX_ES_OUTPUT) && !(filter->flags & DMX_OUTPUT_RAW_MODE))
                        read_len = NON_SEC_BUF_HEADER_SIZE;
                    else
                        read_len = SEC_BUF_HEADER_SIZE;

                    len = read(filter->fd, sec_buf, read_len);
                    if (len <= 0)
                    {
                        GST_DEBUG("read demux filter[%d] failed (%s) %d", fids[i], strerror(errno), errno);
                    }
                }
                g_mutex_unlock(&dmx->lock);
                GST_DEBUG("tid[%x] ch[%d] %x bytes", sec_buf[0], fids[i], len);
                if (len > 0 && filter->cb)
                {
                    filter->cb(filter->dev_no, fids[i], (const guint *)sec_buf, len, filter->user_data);
                }
            }
        }
    }

    if (sec_buf)
    {
        free(sec_buf);
    }

    return NULL;
}

static dvb_dmx_filter_t *dmx_get_filter(dvb_dmx_t *dev, gint fhandle)
{
    if (fhandle >= DMX_FILTER_COUNT)
    {
        GST_DEBUG("wrong filter no");
        return NULL;
    }

    if (!dev->filter[fhandle].used)
    {
        GST_DEBUG("filter %d not allocated", fhandle);
        return NULL;
    }
    return &dev->filter[fhandle];
}

/**\brief dmx device init, creat dmx thread
 * \param dmx device number
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_open(gint dev_no)
{
    dvb_dmx_t *dev = NULL;
    char node[32] = {0};
    char node2[20] = {0};

    if (dmx_get_dev(dev_no, &dev))
        return DVB_FAILURE;

    if (g_atomic_int_get(&dev->running))
    {
        GST_DEBUG("dmx already initialized");
        return DVB_FAILURE;
    }

    set_dmx_source();

    snprintf(node, sizeof(node), "/dev/dvb0.dvr%d", dev_no);
    snprintf(node2, sizeof(node2), "/dev/dvb0.demux%d", dev_no);
    dev->dmx_src_fd = open(node, O_WRONLY);
    dev->dmx_fd = open(node2, O_RDONLY);
    if (dev->dmx_src_fd < 0 || dev->dmx_fd < 0)
    {
        GST_DEBUG("open dmx device fail");
        return DVB_FAILURE;
    }

    dev->dev_no = dev_no;
    g_atomic_int_set(&dev->running, 1);

    gst_amldmxwrap_start_fetch_data(dev_no);

    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_start_fetch_data(gint dev_no)
{
    dvb_dmx_t *dev = NULL;

    if (dmx_get_dev(dev_no, &dev))
        return DVB_FAILURE;

    if (!dev->task)
    {
        dev->task = gst_task_new((GstTaskFunction)gst_amldmxwrap_data_thread, dev, NULL);
        g_rec_mutex_init(&dev->tlock);
        g_mutex_init(&dev->lock);
        gst_task_set_lock(dev->task, &dev->tlock);
    }

    gst_task_start(dev->task);

    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_pause_fetch_data(gint dev_no)
{
    dvb_dmx_t *dev = NULL;

    if (dmx_get_dev(dev_no, &dev))
        return DVB_FAILURE;

    if (!dev->task)
        return DVB_FAILURE;

    gst_task_pause(dev->task);

    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_set_flushing(gint dev_no, gboolean flushing)
{
    dvb_dmx_t *dev = NULL;

    if (dmx_get_dev(dev_no, &dev))
        return DVB_FAILURE;

    if (flushing)
        g_atomic_int_set(&dev->running, 0);
    else
        g_atomic_int_set(&dev->running, 1);

    return DVB_SUCCESS;
}

/**\brief set dmx config
 * \param dmx device number
 * \param is secure scenes
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_set_config(gint dev_no, gboolean is_secure)
{
    dvb_dmx_t *dev = NULL;
    DVB_RESULT dmxret = DVB_SUCCESS;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("demux allocate failed, wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    if (is_secure)
    {
        dmxret |= ioctl(dev->dmx_src_fd, DMX_SET_INPUT, INPUT_LOCAL_SEC);
        dmxret |= ioctl(dev->dmx_fd, DMX_SET_INPUT, INPUT_LOCAL_SEC);
        dmxret |= ioctl(dev->dmx_fd, DMX_SET_HW_SOURCE, DMA_0);
    }
    else
    {
        dmxret |= ioctl(dev->dmx_src_fd, DMX_SET_INPUT, INPUT_LOCAL);
        dmxret |= ioctl(dev->dmx_fd, DMX_SET_INPUT, INPUT_LOCAL);
        dmxret |= ioctl(dev->dmx_fd, DMX_SET_HW_SOURCE, DMA_0);
    }

    return dmxret;
}

/**\brief write data into dmx
 * \param dmx device number
 * \param data data ptr
 * \param size data size
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_write(gint dev_no, gchar *data, guint size)
{
    dvb_dmx_t *dev = NULL;
    DVB_RESULT dmxret = DVB_SUCCESS;
    gchar *offset = 0;
    ssize_t actual_write = 0;
    ssize_t remaining = 0;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("demux allocate failed, wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    offset = data;
    remaining = size;
    GST_DEBUG("try to write ts data into DMX. offset:%p, size:%d", offset, remaining);
    while (remaining)
    {
        actual_write = write(dev->dmx_src_fd, (const void *)offset, (size_t)remaining);
        if (actual_write == remaining)
        {
            GST_DEBUG("write ts data done.");
            remaining = 0;
            dmxret = DVB_SUCCESS;
            break;
        }
        else if (actual_write == -1)
        {
            GST_DEBUG("write ts data error. It is likely to be caused by insufficient es buffer. Need sleep and try again");
            g_usleep(50 * 1000);
        }
        else if (actual_write < remaining)
        {
            remaining -= actual_write;
            offset += actual_write;
            GST_DEBUG("write ts data continue. It is likely to be caused by insufficient es buffer. Need sleep and try again");
            g_usleep(50 * 1000);
        }
        else if (actual_write > remaining)
        {
            GST_ERROR("write ts data out of bounds. Meet real errors");
            dmxret = DVB_FAILURE;
            break;
        }
    }
    GST_DEBUG("write return:%d", dmxret);
    return dmxret;
}

/**\brief allocate dmx filter
 * \param dmx device number
 * \param get dmx filter index
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_alloc_filter(gint dev_no, gint *fhandle)
{
    gint fd;
    gint fid;
    dvb_dmx_filter_t *filter = NULL;
    char dev_name[32];

    dvb_dmx_t *dev = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("demux allocate failed, wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);
    filter = &dev->filter[0];
    for (fid = 0; fid < DMX_FILTER_COUNT; fid++)
    {
        if (!filter[fid].used)
        {
            break;
        }
    }

    if (fid >= DMX_FILTER_COUNT)
    {
        GST_DEBUG("filter id:%d, have no filter to alloc", fid);
        g_mutex_unlock(&dev->lock);
        return DVB_FAILURE;
    }

    memset(dev_name, 0, sizeof(dev_name));
    sprintf(dev_name, "/dev/dvb0.demux%d", dev_no);
    fd = open(dev_name, O_RDWR);
    if (fd == -1)
    {
        GST_DEBUG("cannot open \"%s\" (%s)", dev_name, strerror(errno));
        g_mutex_unlock(&dev->lock);
        return DVB_FAILURE;
    }

    memset(&filter[fid], 0, sizeof(dvb_dmx_filter_t));
    filter[fid].dev_no = dev_no;
    filter[fid].fd = fd;
    filter[fid].used = 1;
    *fhandle = fid;

    g_mutex_unlock(&dev->lock);

    // GST_DEBUG("fhandle = %d", fid);
    return DVB_SUCCESS;
}

/**\brief set demux section filter
 * \param dmx device number
 * \param dmx filter index
 * \param dmx section filter param
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_sec_secfilter(gint dev_no, gint fhandle, const struct dmx_sct_filter_params *params)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("Wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    if (!params)
        return DVB_FAILURE;

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter)
    {
        if (ioctl(filter->fd, DMX_STOP, 0) < 0)
        {
            GST_DEBUG("dmx stop filter failed error:%s", strerror(errno));
            ret = DVB_FAILURE;
        }
        else if (ioctl(filter->fd, DMX_SET_FILTER, params) < 0)
        {
            GST_DEBUG("set filter failed error:%s", strerror(errno));
            ret = DVB_FAILURE;
        }
    }

    g_mutex_unlock(&dev->lock);
    // GST_DEBUG("pid = %x", params->pid);
    return ret;
}

/**\brief set demux pes filter
 * \param dmx device number
 * \param dmx filter index
 * \param dmx pes filter param
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_set_pesfilter(gint dev_no, gint fhandle, const struct dmx_pes_filter_params *params)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_ERROR("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    if (!params)
        return DVB_FAILURE;

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter)
    {
        ret = DVB_FAILURE;
        if (ioctl(filter->fd, DMX_STOP, 0) < 0)
        {
            GST_ERROR("stopping demux filter fails with errno:%d(%s)", errno, strerror(errno));
        }
        else if (fcntl(filter->fd, F_SETFL, O_NONBLOCK) < 0)
        {
            GST_ERROR("setting filter non-block flag fails with errno:%d(%s)", errno, strerror(errno));
        }
        else if (ioctl(filter->fd, DMX_SET_PES_FILTER, params) < 0)
        {
            GST_ERROR("setting PES filter fails with errno:%d(%s)", errno, strerror(errno));
        }
        else
        {
            filter->flags = params->flags;
            ret = DVB_SUCCESS;
        }
    }

    g_mutex_unlock(&dev->lock);
    // GST_DEBUG("pid = %x", params->pid);
    return ret;
}

/**\brief set demux filter buffer
 * \param dmx device number
 * \param dmx filter index
 * \param dmx filter buffer size
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_set_bufsize(gint dev_no, gint fhandle, gint size)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter)
    {
        if (ioctl(filter->fd, DMX_SET_BUFFER_SIZE, size) < 0)
        {
            GST_DEBUG("set buf size failed error:%s", strerror(errno));
            ret = DVB_FAILURE;
        }
    }

    g_mutex_unlock(&dev->lock);
    return ret;
}

/**\brief free demux filter
 * \param dmx device number
 * \param dmx filter index
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_free_filter(gint dev_no, gint fhandle)
{
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter)
    {
        filter->need_free = 1;
    }

    g_mutex_unlock(&dev->lock);

    // GST_DEBUG("fhandle = %d", fhandle);
    return DVB_SUCCESS;
}

/**\brief start demux filter
 * \param dmx device number
 * \param dmx filter index
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_start_filter(gint dev_no, gint fhandle)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    GST_DEBUG("trace in start filter:%d", fhandle);
    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter && !filter->enable)
    {
        if (ioctl(filter->fd, DMX_START, 0) < 0)
        {
            GST_DEBUG("dmx start filter failed error:%s", strerror(errno));
            ret = DVB_FAILURE;
        }
        else
        {
            filter->enable = 1;
        }
    }

    g_mutex_unlock(&dev->lock);

    gst_aml_dmabuf_mgr_set_filter_info(fhandle, filter->fd, 0);

    return ret;
}

/**\brief stop demux filter
 * \param dmx device number
 * \param dmx filter index
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_stop_filter(gint dev_no, gint fhandle)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);

    filter = dmx_get_filter(dev, fhandle);
    if (filter && filter->enable)
    {
        if (ioctl(filter->fd, DMX_STOP, 0) < 0)
        {
            GST_DEBUG("dmx stop filter failed error:%s", strerror(errno));
            ret = DVB_FAILURE;
        }
        else
        {
            filter->enable = 0;
        }
    }

    g_mutex_unlock(&dev->lock);

    gst_aml_dmabuf_mgr_set_filter_info(fhandle, filter->fd, 1);

    return ret;
}

/**\brief set demux callback
 * \param dmx device number
 * \param dmx filter index
 * \param dmx filter callback
 * \param dmx filter callback param
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_set_cb(gint dev_no, gint fhandle, gst_amldmxwrap_data_cb cb, void *user_data)
{
    DVB_RESULT ret = DVB_SUCCESS;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_DEBUG("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);
    filter = dmx_get_filter(dev, fhandle);
    if (filter)
    {
        filter->cb = cb;
        filter->user_data = user_data;
    }
    else
    {
        ret = DVB_FAILURE;
    }

    g_mutex_unlock(&dev->lock);

    // GST_DEBUG("ret = %d", ret);
    return ret;
}

/**\brief dmx device uninit, destroy dmx thread
 * \param dmx device number
 * \return DVB_SUCCESS On success, DVB_FAILURE on error.
 */
DVB_RESULT gst_amldmxwrap_close(gint dev_no)
{
    gint i;
    gint open_count = 0;
    dvb_dmx_t *dev = NULL;
    dvb_dmx_filter_t *filter = NULL;
    DVB_RESULT ret = DVB_SUCCESS;

    GST_INFO("trace in");

    if (dmx_get_dev(dev_no, &dev))
    {
        GST_ERROR("wrong dmx device no %d", dev_no);
        return DVB_FAILURE;
    }

    g_mutex_lock(&dev->lock);

    for (i = 0; i < DMX_FILTER_COUNT; i++)
    {
        filter = &dev->filter[i];
        if (filter->used && filter->dev_no == dev_no)
        {
            if (filter->enable)
            {
                if (ioctl(filter->fd, DMX_STOP, 0) < 0)
                {
                    GST_ERROR("fails to stop filter. fd:%d", filter->fd);
                    ret = DVB_FAILURE;
                }
            }
            close(filter->fd);
        }
        else if (filter->used)
        {
            open_count++;
        }
    }

    if (open_count == 0)
    {
        g_atomic_int_set(&dev->running, 0);

        gst_task_stop(dev->task);
        g_rec_mutex_lock(&dev->tlock);
        g_rec_mutex_unlock(&dev->tlock);
        gst_task_join(dev->task);
        gst_object_unref(dev->task);
        g_rec_mutex_clear(&dev->tlock);
        g_mutex_clear(&dev->lock);
    }

    if (dev->dmx_fd >= 0)
    {
        close(dev->dmx_fd);
        dev->dmx_fd = -1;
    }

    if (dev->dmx_src_fd >= 0)
    {
        close(dev->dmx_src_fd);
        dev->dmx_src_fd = -1;
    }

    GST_INFO("trace out");
    return ret;
}

DVB_RESULT gst_amldmxwrap_get_es(gint dev_no, gint fhandle, guint8 *data, guint size)
{
    dvb_dmx_t *dmx = NULL;
    dvb_dmx_filter_t *filter = NULL;
    gint ret = 0;
    struct pollfd fd[1];

    GST_DEBUG("need read size:%d", size);

    if (dmx_get_dev(dev_no, &dmx))
    {
        GST_ERROR("get dmx error");
        return DVB_FAILURE;
    }

    filter = &dmx->filter[fhandle];
    if (!filter || !filter->enable || !filter->used || filter->need_free)
    {
        GST_ERROR("ch[%d] not used, not read", fhandle);
        return DVB_FAILURE;
    }

    while (1)
    {
        memset(fd, 0, sizeof(struct pollfd));
        fd[0].events = POLLIN | POLLERR;
        fd[0].fd = filter->fd;
        ret = poll(fd, 1, DMX_POLL_TIMEOUT);
        if (ret <= 0)
        {
            if (!g_atomic_int_get(&dmx->running))
            {
                GST_DEBUG("flushing");
                return DVB_SUCCESS;
            }

            GST_DEBUG("poll ret:%d timeout. need continue", ret);
            continue;
        }

        if (fd[0].revents & (POLLIN | POLLERR))
        {
            // pthread_mutex_lock(&dmx->lock);

            gint tmp = 0;
            guint read_len = 0;
            guint left = size;
            do
            {
                tmp = read(filter->fd, data + read_len, left);
                if (tmp <= 0)
                {
                    GST_ERROR("read demux filter[%d] failed (%s) %d", fhandle, strerror(errno), errno);
                    return DVB_FAILURE;
                }
                read_len += tmp;
                left -= tmp;
                GST_DEBUG("total:%d, read:%d left:%d", size, read_len, left);
            } while (left > 0);

            GST_DEBUG("read demux filter[%d] ok", fhandle);

            // pthread_mutex_unlock(&dmx->lock);
            break;
        }
    }

    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_get_filters_mem_info(gint dev_no, struct dmx_filter_mem_info *info_all)
{
    dvb_dmx_t *dmx = NULL;

    if (dmx_get_dev(dev_no, &dmx))
    {
        GST_ERROR("get dmx error");
        return DVB_FAILURE;
    }

    memset(info_all, 0, sizeof(struct dmx_filter_mem_info));
    if (ioctl(dmx->dmx_fd, DMX_GET_FILTER_MEM_INFO, info_all) < 0)
    {
        GST_ERROR("dmx filter get mem info fail(%s) %d", strerror(errno), errno);
        return DVB_FAILURE;
    }

    GST_DEBUG("got filters num:%d", info_all->filter_num);
    if (info_all->filter_num > 0)
    {
        for (int i = 0; i < info_all->filter_num; i++)
        {
            struct dmx_mem_info *test_info = &info_all->info[i].filter_info;
            GST_DEBUG("filter[idx:%d type:%d pid:%d] dmx_total_size:%d, dmx_buf_phy_start:0x%x, dmx_free_size:%d, dvb_core_total_size:%d, dvb_core_free_size:%d",
                      i, info_all->info[i].type, info_all->info[i].pid,
                      test_info->dmx_total_size, test_info->dmx_buf_phy_start, test_info->dmx_free_size, test_info->dvb_core_total_size, test_info->dvb_core_free_size);
        }
    }

    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_set_video_filters_es_buf_size(guint size)
{
    const char *path = "/sys/module/dvb_demux/parameters/video_buf_size";
    gint fd;
    gchar cmd[128] = {0};

    fd = open(path, O_CREAT | O_RDWR | O_TRUNC, 0644);
    if (fd < 0)
        return DVB_FAILURE;

    sprintf(cmd, "%d", size);
    write(fd, cmd, strlen(cmd));
    close(fd);
    return DVB_SUCCESS;
}

DVB_RESULT gst_amldmxwrap_set_filter_ring_buf_size(gint dev_no, gint fhandle, guint size)
{
    dvb_dmx_t *dmx = NULL;
    dvb_dmx_filter_t *filter = NULL;

    GST_DEBUG("trace in");

    if (dmx_get_dev(dev_no, &dmx))
    {
        GST_ERROR("get dmx error");
        return DVB_FAILURE;
    }

    filter = dmx_get_filter(dmx, fhandle);
    if (!filter)
    {
        GST_ERROR("get filter error");
        return DVB_FAILURE;
    }
    if (ioctl(filter->fd, DMX_SET_BUFFER_SIZE, size) < 0)
    {
        GST_ERROR("dmx filter get mem info fail(%s) %d", strerror(errno), errno);
        return DVB_FAILURE;
    }
    GST_DEBUG("trace out set size:%d ok", size);
    return DVB_SUCCESS;
}