source: svn/newcon3bcm2_21bu/BSEAV/lib/utils/bsink_playback.c

/***************************************************************************
 *     Copyright (c) 2007 Broadcom Corporation
 *     All Rights Reserved
 *     Confidential Property of Broadcom Corporation
 *
 *  THIS SOFTWARE MAY ONLY BE USED SUBJECT TO AN EXECUTED SOFTWARE LICENSE
 *  AGREEMENT  BETWEEN THE USER AND BROADCOM.  YOU HAVE NO RIGHT TO USE OR
 *  EXPLOIT THIS MATERIAL EXCEPT SUBJECT TO THE TERMS OF SUCH AN AGREEMENT.
 *
 * $brcm_Workfile: bsink_playback.c $
 * $brcm_Revision: 6 $
 * $brcm_Date: 11/24/10 2:16p $
 *
 * Module Description:
 *
 * ATOM playback data sink
 *
 * Revision History:
 *
 * $brcm_Log: /BSEAV/lib/utils/bsink_playback.c $
 * 
 * 6   11/24/10 2:16p vsilyaev
 * SW35230-2195: Added option to seed timing on first read (helps when
 * playback started in paused state)
 * 
 * 5   11/23/10 8:39p vsilyaev
 * SW35230-2195: Added compile time option to capture and restore stream
 * data with relative timing
 * 
 * 4   6/24/08 5:53p erickson
 * PR44070: added fflush for  v
 *
 * 3   9/24/07 5:35p gmohile
 * PR 25109: Dump PES out
 *
 * 2   9/20/07 7:16p gmohile
 * PR 25109: Fix segfaults
 *
 * 1   8/2/07 3:57p vsilyaev
 * PR 33751: Playback sink (destination) in SPF framework
 *
 *
 ***************************************************************************/
#include "bstd.h"
#include "bsink_playback.h"
#include "bkni.h"

#define BDBG_MSG_TRACE(x) /* BDBG_MSG(x) */
BDBG_MODULE(bsink_playback);

/* only one should be enabled at the time, WRITE writes data to file and READ reads this data leter on */
/* #define WRITE_TIMED_DATA 1 */
/* #define READ_TIMED_DATA 1 */

/* #define DUMP_PES_OUT 1 */

#ifdef DUMP_PES_OUT
#include <stdio.h>
FILE *fp;
#endif

#if WRITE_TIMED_DATA  || READ_TIMED_DATA
#include "bsink_playback_debug.h"
#include <stdio.h>
#include <sys/time.h>

static FILE *
b_sink_playback_open_file(const char *mode, unsigned instance, unsigned *count)
{
    FILE *f;
    char name[64];

    BKNI_Snprintf(name, sizeof(name), "timed_data_%u_%u.dat",  instance, *count);
    BDBG_WRN(("using dump file '%s'", name));
    f = fopen(name, mode);
    BDBG_ASSERT(f);
    (*count)++;
    return f;
}

static long
b_sink_playback_dump_timediff(const struct timeval *future, const struct timeval *past)
{
    return 1000*(future->tv_sec - past->tv_sec) + (future->tv_usec - past->tv_usec)/1000;
}
#endif /* WRITE_TIMED_DATA  || READ_TIMED_DATA */

#if READ_TIMED_DATA
#include "balloc.h"
balloc_iface_t bsink_dump_reader_allocator = NULL;

typedef struct b_sink_playback_read {
    FILE *fp;
    bool get_time;
    unsigned count;
    unsigned instance;
    struct timeval start_time;
} b_sink_playback_read;

static void 
b_sink_playback_dump_init(b_sink_playback_read *reader)
{
    static int bsink_instance=0;

    reader->fp = NULL;
    reader->count = 0;
    reader->instance = bsink_instance;
    bsink_instance++;
    return;
}

static void 
b_sink_playback_dump_start(b_sink_playback_read *reader)
{
    reader->fp = b_sink_playback_open_file("rb", reader->instance, &reader->count);
#if 1
    reader->get_time = true;
#else
    reader->get_time = false;
    gettimeofday(&reader->start_time, NULL);
#endif
    return;
}

static void 
b_sink_playback_dump_stop(b_sink_playback_read *reader)
{
    fclose(reader->fp);
    reader->fp = NULL;
}

static void
b_sink_reader_free(batom_t atom, void *user)
{
    batom_t src_atom = *(batom_t *)user;
        unsigned nvecs, i;
    const batom_vec *vecs;

    BDBG_MSG(("b_sink_reader_free: %#lx %#lx", (unsigned long)atom, (unsigned long)src_atom));
        vecs = batom_get_vectors(atom, &nvecs);
    for(i=0;i<nvecs;i++) {
        bsink_dump_reader_allocator->bmem_free(bsink_dump_reader_allocator, vecs[i].base);
    }
    batom_release(src_atom);
    return;
}

static const batom_user b_sink_reader = {
        b_sink_reader_free,
        sizeof(batom_t)
};

static batom_t
b_sink_playback_dump_data(b_sink_playback_read *reader, batom_t atom, batom_factory_t factory)
{
        unsigned nvecs, i;
    int rc;
    batom_vec *dst_vec;
    batom_vec dst_static_vec[4];
    bsink_playback_packet packet;
    struct timeval now;
    uint32_t time_diff;

    BDBG_ASSERT(bsink_dump_reader_allocator);
    if(reader->get_time) {
        reader->get_time = false;
        gettimeofday(&reader->start_time, NULL);
    }
    gettimeofday(&now, NULL);
    for(;;) {
        rc = fread(&packet, sizeof(packet), 1, reader->fp);
        if(rc==1) {
            BDBG_ASSERT(packet.sync == BSINK_PLAYBACK_SYNC);
            break;
        }
        BDBG_ASSERT(rc==0);
        BDBG_WRN(("end of source data, loop"));
        fseek(reader->fp, 0, SEEK_SET);
        gettimeofday(&reader->start_time, NULL);
    }
    time_diff = b_sink_playback_dump_timediff(&now, &reader->start_time);
    if(packet.timestamp > time_diff) {
        BKNI_Sleep(packet.timestamp - time_diff);
    }
    nvecs = packet.sequence+1;

    if(nvecs<sizeof(dst_static_vec)/sizeof(*dst_static_vec)) {
        dst_vec = dst_static_vec;
    } else {
        dst_vec = BKNI_Malloc(sizeof(*dst_vec)*nvecs);
        BDBG_ASSERT(dst_vec);
    }
    for(i=0;;) {
        dst_vec[i].dirty = true;
        dst_vec[i].unused_flags = 0;
        dst_vec[i].len = packet.length;
        dst_vec[i].base = bsink_dump_reader_allocator->bmem_alloc(bsink_dump_reader_allocator, packet.length);
        BDBG_ASSERT(dst_vec[i].base);
        rc = fread(dst_vec[i].base, packet.length, 1, reader->fp);
        BDBG_ASSERT(rc==1);
        i++;
        if(i>=nvecs) {
            break;
        }
        rc = fread(&packet, sizeof(packet), 1, reader->fp);
        BDBG_ASSERT(rc==1);
        BDBG_ASSERT(packet.sync == BSINK_PLAYBACK_SYNC);
    }
    atom = batom_from_vectors(factory, dst_vec, nvecs, &b_sink_reader, &atom);
    BDBG_ASSERT(atom);
    if(dst_vec != dst_static_vec) {
        BKNI_Free(dst_vec);
    }
    return atom;
}
#elif WRITE_TIMED_DATA  
typedef struct b_sink_playback_dump {
    FILE *fp;
    unsigned count;
    unsigned instance;
    struct timeval start_time;
} b_sink_playback_dump;

static void 
b_sink_playback_dump_init(b_sink_playback_dump *writer)
{
    static int bsink_instance=0;

    writer->fp = NULL;
    writer->count = 0;
    writer->instance = bsink_instance;
    bsink_instance++;
    return;
}

static void 
b_sink_playback_dump_start(b_sink_playback_dump *writer)
{
    writer->fp = b_sink_playback_open_file("wb", writer->instance, &writer->count);
    gettimeofday(&writer->start_time, NULL);
    return;
}

static void 
b_sink_playback_dump_stop(b_sink_playback_dump *writer)
{
    fclose(writer->fp);
    writer->fp = NULL;
}


static batom_t
b_sink_playback_dump_data(b_sink_playback_dump *writer, batom_t atom, batom_factory_t factory)
{
        unsigned nvecs, i;
    const batom_vec *vecs;
    bsink_playback_packet packet;
    struct timeval now;

    BSTD_UNUSED(factory);
    gettimeofday(&now, NULL);
    packet.sync = BSINK_PLAYBACK_SYNC;
        vecs = batom_get_vectors(atom, &nvecs);
    packet.timestamp = b_sink_playback_dump_timediff(&now, &writer->start_time);
    for(i=0;i<nvecs;i++) {
        packet.sequence =  nvecs - i - 1;
        packet.length = vecs[i].len;
        fwrite(&packet, sizeof(packet), 1, writer->fp);
        fwrite(vecs[i].base, 1, vecs[i].len, writer->fp);
    }
    fflush(writer->fp);
    return atom;
}
#else
#define b_sink_playback_dump_init(writer)
#define b_sink_playback_dump_start(writer)
#define b_sink_playback_dump_stop(writer)
#define b_sink_playback_dump_data(writer, atom, factory) (atom)
#endif /* WRITE_TIMED_DATA */


BDBG_OBJECT_ID(bsink_playback_t);


#define B_SINK_PLAYBACK_NENTRIES        32
struct bsink_playback {
        BDBG_OBJECT(bsink_playback_t)
        bsink_playback_cfg cfg;
        struct {
                batom_t atom;
                unsigned vec;
        } in, out;
        batom_pipe_t out_pipe;
        unsigned nfreedesc;
        unsigned nplaydesc;
        BPVRlib_Feed_OffsetEntry entries[B_SINK_PLAYBACK_NENTRIES];
    batom_factory_t factory;
#if READ_TIMED_DATA
    struct b_sink_playback_read dumper;
#elif WRITE_TIMED_DATA 
    struct b_sink_playback_dump dumper;
#endif
};


void
bsink_playback_default_cfg(bsink_playback_cfg *cfg)
{
        cfg->addr_to_offset = NULL;
        cfg->cache_flush = NULL;
        return;
}

static void
b_sink_playback_reset(bsink_playback_t sink)
{
        sink->in.atom = NULL;
        sink->in.vec = 0;
        sink->out.atom = NULL;
        sink->out.vec = 0;
        sink->nfreedesc = 0;
        sink->nplaydesc = 0;
        return;
}

bsink_playback_t
bsink_playback_create(batom_factory_t factory, const bsink_playback_cfg *cfg)
{
        bsink_playback_t sink;

        BDBG_ASSERT(factory);
        BDBG_ASSERT(cfg);
        if(cfg->addr_to_offset == NULL || cfg->cache_flush==NULL) {
                BDBG_ERR(("bsink_playback_create: cfg.addr_to_offset and cfg.cache_flush shall be specified"));
                goto err_cfg;
        }
        sink = BKNI_Malloc(sizeof(*sink));
        if(!sink) {
                BDBG_ERR(("bsink_playback_create: can't allocate %u bytes", sizeof(*sink)));
                goto err_alloc;
        }
        BDBG_OBJECT_INIT(sink, bsink_playback_t);
        sink->cfg = *cfg;
    sink->factory = factory;
        sink->out_pipe = batom_pipe_create(factory);
        if(!sink->out_pipe) {
                goto err_out_pipe;
        }
    b_sink_playback_dump_init(&sink->dumper);
        b_sink_playback_reset(sink);
        return sink;

err_out_pipe:
        BKNI_Free(sink);
err_alloc:
err_cfg:
        return NULL;
}

void
bsink_playback_destroy(bsink_playback_t sink)
{
        BDBG_OBJECT_ASSERT(sink, bsink_playback_t);
        batom_pipe_destroy(sink->out_pipe);
        BDBG_OBJECT_DESTROY(sink, bsink_playback_t);
        BKNI_Free(sink);
        return;
}

BERR_Code
bsink_playback_start(bsink_playback_t sink)
{
        BERR_Code rc;
        BPVRlib_Feed_Status status;

        BDBG_OBJECT_ASSERT(sink, bsink_playback_t);
        b_sink_playback_reset(sink);
        rc = BPVRlib_Feed_Start(sink->cfg.feed);
        if(rc!=BERR_SUCCESS) {
                rc = BERR_TRACE(rc);
                goto error;
        }
        BPVRlib_Feed_GetStatus(sink->cfg.feed, &status);
        sink->nfreedesc = status.freeDesc;
        sink->nplaydesc = status.freeDesc;
#ifdef DUMP_PES_OUT
        fp=fopen("media_out.mpg", "wb");
#endif
    b_sink_playback_dump_start(&sink->dumper);
error:
        BDBG_MSG(("bsink_playback_start: %#lx %#x", (unsigned long)sink, (unsigned)rc));
        return rc;
}

void
bsink_playback_stop(bsink_playback_t sink)
{
        BDBG_OBJECT_ASSERT(sink, bsink_playback_t);
        BDBG_MSG(("bsink_playback_stop: %#lx", (unsigned long)sink));

#if 0
        if(sink->in.atom) {
                batom_release(sink->in.atom);
                sink->in.atom = NULL;
        }
#endif
        if(sink->out.atom) {
                batom_release(sink->out.atom);
                sink->out.atom = NULL;
        }
        batom_pipe_flush(sink->out_pipe);
        BPVRlib_Feed_Stop(sink->cfg.feed);
    b_sink_playback_dump_stop(&sink->dumper);
#ifdef DUMP_PES_OUT
        fclose(fp);
#endif
        return;
}


size_t
bsink_playback_feed(bsink_playback_t sink, batom_pipe_t in_pipe)
{
        BERR_Code rc;
        unsigned i;
        unsigned nvecs, vec, last_vec;
        size_t ncompleted;
        size_t nentries;
        batom_t atom;
        size_t len=0;
        BPVRlib_Feed_OffsetEntry *entries = sink->entries;

        BDBG_OBJECT_ASSERT(sink, bsink_playback_t);
        BDBG_ASSERT(in_pipe);
        BDBG_MSG_TRACE(("bsink_playback_feed>: %#lx %#lx",  (unsigned long)sink, (unsigned long)in_pipe));

        /* step 1. get number of recycled vectors */
        rc = BPVRlib_Feed_GetCompleted(sink->cfg.feed, &ncompleted);
        if(rc!=BERR_SUCCESS) {
                rc = BERR_TRACE(rc);
                goto error;
        }
        BDBG_MSG_TRACE(("bsink_playback_feed: %#lx completed:%u",  (unsigned long)sink, ncompleted));
        /* step 2. recycle entries from out_pipe */
        atom = sink->out.atom; /* load state */
        vec = sink->out.vec;
        for(i=ncompleted;i>0;) {
                if(!atom) {
                        vec = 0;
                        atom = batom_pipe_pop(sink->out_pipe);
                        if(!atom) {
                                BDBG_ERR(("bsink_playback_feed: %#lx underflow left:%u", (unsigned long)sink, i));
                                break;
                        }
                        BDBG_MSG_TRACE(("bsink_playback_feed: %#lx recycling:%#lx",  (unsigned long)sink, atom));
                }
                nvecs = batom_get_nvecs(atom);
                BDBG_ASSERT(nvecs>vec);
                last_vec = vec;
                vec+=i;
                if(vec < nvecs) {
                        break;
                }
                /* completed entire atom */
                BDBG_ASSERT(i>=(nvecs-last_vec));
                i -= nvecs - last_vec;
                vec = 0;
                batom_release(atom);
                atom=NULL;
        }
        sink->out.vec = vec; /* save state for the next step */
        sink->out.atom = atom;

        /* step 3. Update number of free descriotrs */
        sink->nfreedesc += ncompleted;
        BDBG_ASSERT(sink->nfreedesc<=sink->nplaydesc);

        vec = sink->in.vec; /* load state */
        atom = sink->in.atom;

        /* step 4. Copy data from in_pipe into the BPVRlib_Feed entries */
        do {
                nentries = B_SINK_PLAYBACK_NENTRIES;
                if(nentries>sink->nfreedesc) {
                        nentries = sink->nfreedesc;
                }
                BDBG_MSG_TRACE(("bsink_playback_feed: %#lx free:%u",  (unsigned long)sink, nentries));
                for(i=0;i<nentries;) {
                        const batom_vec *vecs;
                        if(!atom) {
                                vec = 0;
                                atom = batom_pipe_pop(in_pipe);
                                if(!atom) {
                                        break;
                                }
                atom = b_sink_playback_dump_data(&sink->dumper, atom, sink->factory);
                                BDBG_MSG_TRACE(("bsink_playback_feed: %#lx feeding %#lx",  (unsigned long)sink, atom));
                                batom_pipe_push(sink->out_pipe, atom);
                        }
                        vecs = batom_get_vectors(atom, &nvecs);
                        BDBG_ASSERT(nvecs>vec);
                        last_vec = vec + (nentries-i);
                        if( last_vec > nvecs) {
                                last_vec = nvecs;
                        }
                        BDBG_ASSERT(i+(last_vec-vec)<=nentries);
                        /* copy vectors to the feed entries */
                        for(;vec<last_vec;vec++,i++) {
                                len += vecs[vec].len;
                                entries[i].len = vecs[vec].len;
                                if(vecs[vec].dirty) {
                                        sink->cfg.cache_flush(vecs[vec].base, vecs[vec].len);
                                }
                                entries[i].offset = sink->cfg.addr_to_offset(vecs[vec].base);
#ifdef DUMP_PES_OUT
                                fwrite(vecs[vec].base, vecs[vec].len, 1, fp);
                                fflush(fp);
#endif
                        }
                        if(vec!=nvecs) { /* some vectors are left in the atom */
                                BDBG_ASSERT(i==nentries);
                                break;
                        }
                        /* copied entire vector */
                        atom = NULL;
                }
                sink->in.vec = vec;
                sink->in.atom = atom; /* save state for the next step */
                if(i==0) {
                        break; /* nothing to send */
                }
                BDBG_ASSERT(sink->nfreedesc>=i);
                sink->nfreedesc -= i;
                rc = BPVRlib_Feed_AddOffsetEntries(sink->cfg.feed, entries, i, &ncompleted);
                if(rc!=BERR_SUCCESS) {
                        rc = BERR_TRACE(rc);
                        goto error;
                }
                if(i!=ncompleted) {
                        BDBG_ERR(("bsink_playback_feed: %#lx BPVRlib_Feed haven't consumed expected number of entries %u:%u", (unsigned long)sink, i, ncompleted));
                        goto error;
                }
        } while(i==B_SINK_PLAYBACK_NENTRIES); /* repeat until we send less than complete chunk, in which case it means either BPVRlib_Feed is full or in.pipe is empty */
error:
        BDBG_MSG_TRACE(("bsink_playback_feed<: %#lx  ->%u",  (unsigned long)sink, len));
        return len;
}


void
bsink_playback_get_status(bsink_playback_t sink, bsink_playback_status *status)
{
        BDBG_OBJECT_ASSERT(sink, bsink_playback_t);
        /* sink is idle if no atom is borrowed and feed is idle (e.g. number of freedescriptors is equal to total number of descriptors */
        status->idle = sink->in.atom==NULL && sink->nfreedesc == sink->nplaydesc;
        BDBG_MSG_TRACE(("bsink_playback_status: %#lx %s", (unsigned long)sink, status->idle?"IDLE":""));
        return;
}