stmr_task.c

Go to the documentation of this file.

/*******************************************************************************
 * Copyright © 2014 The DTVKit Open Software Foundation Ltd (www.dtvkit.org)
 * Copyright © 2010 Ocean Blue Software Ltd
 *
 * This file is part of a DTVKit Software Component
 * You are permitted to copy, modify or distribute this file subject to the terms
 * of the DTVKit 1.0 Licence which can be found in licence.txt or at www.dtvkit.org
 *
 * THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND,
 * EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
 *
 * If you or your organisation is not a member of DTVKit then you have access
 * to this source code outside of the terms of the licence agreement
 * and you are expected to delete this and any associated files immediately.
 * Further information on DTVKit, membership and terms can be found at www.dtvkit.org
 *******************************************************************************/
/*---includes for this file--------------------------------------------------*/
#include <assert.h>
#include <string.h>

#include "mh5debug.h"
#include "httptype.h"
#include "stb_os.h"
#include "dvb_ics.h"

#include "stmr_queue.h"
#include "stmr_common.h"

/*---constant definitions for this file--------------------------------------*/

/* Look-ahead time difference for flow control (in milliseconds). This
 * determines how far "stream time" can be ahead of "real time". Larger
 * values introduce risk of overflow, smaller values introduce risks of
 * underflow. 1000ms seems to be a good value for the DTG tests.
 */
#define LOOK_AHEAD_DIFF 1000

/* Look-ahead threshold takes care of "strange" PCR differences, for example
 * if the stream loops (PCR jumps to the initial value), or if two streams
 * were concatenated with completely different PCRs (unpredictable jump).
 * The look-ahead is expected to be somewhere between 0 and LOOK_AHEAD_DIFF
 * (could be slightly negative if the platform blocks while data is played).
 * If it escapes from this range, new values are calculated.
 */
#define LOOK_AHEAD_THRESHOLD 2500

#define PDBG(x) TRACE(TICS, x)
#define DBG(x)


/*---local typedef structs for this file-------------------------------------*/

typedef enum
{
   TASK_RUNNING,
   TASK_STOP_REQUEST,
   TASK_STOPPED
} TaskState;

/*---local function definitions----------------------------------------------*/

/*---global function definitions---------------------------------------------*/
static void TaskMain(void *task_arg);
static void HandleQueueItem(MHEG5QueueItem *item);

/*---local (static) variable declarations for this file----------------------*/
static void *taskSemaphore = NULL;
static void *taskHandle;

static void (*underflowCallback)(U32BIT requestId,U32BIT playoutTime) = NULL;

static volatile TaskState taskState;
static volatile BOOLEAN producerCanSignal = FALSE;


E_MhegErr MHEG5StreamerStartTask(U32BIT stack, U8BIT priority)
{
   E_MhegErr result;
   taskState = TASK_RUNNING;
   taskSemaphore = STB_OSCreateSemaphore();
   if (taskSemaphore == NULL)
   {
      result = MHERR_SYSTEM_RESOURCE_SEMAPHORE;
   }
   else
   {
      /* Initial value for semaphore is 1, so do wait to decrement value to 0 */
      STB_OSSemaphoreWait(taskSemaphore);
      taskHandle = STB_OSCreateTask(TaskMain, NULL, stack, priority, (U8BIT *)"MHST");
      if (taskHandle == NULL)
      {
         result = MHERR_SYSTEM_RESOURCE_TASK;
      }
      else
      {
         result = MHERR_OK;
      }
   }
   return result;
}

void MHEG5StreamerSendSignalToTask(void)
{
   if (taskSemaphore != NULL && producerCanSignal)
   {
      producerCanSignal = FALSE;
      STB_OSSemaphoreSignal(taskSemaphore);
   }
}

void MHEG5StreamerRegisterUnderflowCallback(void (*callback)(U32BIT requestId,U32BIT playoutTime))
{
   underflowCallback = callback;
}

void MHEG5StreamerStopTask(void)
{
   void *semaphore;

   taskState = TASK_STOP_REQUEST;
   MHEG5StreamerSendSignalToTask();

   /* It's possible that the semaphore was signalled by someone else. We
    * have to check the stop conditions and keep checking until the
    * task signals the semaphore to announce that it's stopping.
    */
   while (taskState == TASK_RUNNING || taskState == TASK_STOP_REQUEST)
   {
      /* Just busy wait for the task to acknowledge */
   }
   taskHandle = NULL;

   semaphore = taskSemaphore;
   taskSemaphore = NULL;
   STB_OSDeleteSemaphore(semaphore);
}

/*******************
 * LOCAL FUNCTION   *
 ********************/

static void TaskMain(void *task_arg)
{
   MHEG5QueueItem *item;
   U32BIT lastDownloadId;
   U32BIT lastRequestId;
   U32BIT refItemTime, lastItemTime;
   U32BIT clockTime, refClockTime;
   U32BIT relativeClockTime, relativeItemTime;
   S32BIT lastLookahead;
#ifdef TRACING
   U32BIT prevItemTime = 0;
#endif

   refItemTime = 0;
   lastItemTime = 0;
   relativeItemTime = 0;
   lastLookahead = 0;

   lastDownloadId = 0;
   lastRequestId = 0;

   DBG(("Entering main loop"));
   while (taskState == TASK_RUNNING)
   {
      PDBG(("Waiting for semaphore"));
      producerCanSignal = TRUE;
      STB_OSSemaphoreWait(taskSemaphore);
      producerCanSignal = FALSE;

      /* Reference clock time depends on last look-ahead:
       *
       *   lastLookahead = relativeItemTime - relativeClockTime   (1)
       *   relativeClockTime = clockTime - refClockTime
       *
       * So:
       *
       *   refClockTime = clockTime - relativeClockTime
       *   relativeClockTime = relativeItemTime - lastLookahead
       *
       * And so:
       *
       *   refClockTime = clockTime - (relativeItemTime - lastLookahead)
       *                = lastLookahead + clockTime - relativeItemTime
       *
       * lastLookahead - relativeItemTime = -relativeClockTime (according
       * to (1)). As clockTime - relativeClockTime should be positive,
       * refClockTime should also be positive.
       *
       */

      clockTime = STB_OSGetClockMilliseconds();
      refClockTime = lastLookahead + clockTime - relativeItemTime;
      PDBG(("Recalculating refClockTime: %d + %d - %d = %d",
           lastLookahead, clockTime, relativeItemTime, refClockTime));
      DBG(("taskState = %d", taskState));
      while (taskState == TASK_RUNNING)
      {
         item = MHEG5QueueGetHeadItem();
         DBG(("item = %p [%d]", item, MHEG5QueueIsValidItem(item)));
         if (item == NULL)
         {
            PDBG(("Streaming stopped RID=%d",lastRequestId));
            if (underflowCallback != NULL)
            {
               PDBG(("lla=%d-(%d-%d)",lastLookahead,STB_OSGetClockMilliseconds(),clockTime));
               if (lastLookahead > 0)
               {
                  refClockTime = STB_OSGetClockMilliseconds();
                  clockTime += lastLookahead;
                  if (clockTime > refClockTime)
                  {
                     clockTime -= refClockTime;
                  }
                  else
                  {
                     clockTime = 0;
                  }
               }
               else
               {
                  clockTime = 0;
               }
               underflowCallback(lastRequestId,clockTime);
            }
            break;
         }
         if (!MHEG5QueueIsValidItem(item))
         {
            PDBG(("Invalid item, streaming stopped"));
            break;
         }
         clockTime = STB_OSGetClockMilliseconds();

         if (lastDownloadId == item->downloadId)
         {
            if (refClockTime == 0 && clockTime != 0)
            {
               /* Update first time */
               refClockTime = clockTime;
            }
            if (item->timestamp != 0)
            {
               if (refItemTime == 0)
               {
                  /* Update first item time */
                  refItemTime = item->timestamp;
               }
            #ifdef TRACING
               prevItemTime = lastItemTime;
            #endif
               lastItemTime = item->timestamp;
            }
            if ((clockTime < refClockTime) ||
                (lastItemTime < refItemTime) ||
                (lastLookahead < -LOOK_AHEAD_THRESHOLD) ||
                (lastLookahead > LOOK_AHEAD_THRESHOLD))
            {
               /* Overflow or other issue - reset */
               PDBG(("Resetting PCR monitor - invalid values (%u,%u) (%u,%u) %d prev=%u",
                  clockTime,refClockTime, lastItemTime,refItemTime, lastLookahead, prevItemTime ));
               refClockTime = 0;
               refItemTime = 0;
               lastItemTime = 0;
               lastLookahead = 0;
               relativeItemTime = 0;
            }
         }
         else
         {
            /* New download - reset */
            PDBG(("Resetting PCR monitor RID=%d->%d  DID=%d->%d",lastRequestId,item->requestId,lastDownloadId,item->downloadId));
            PDBG(("Timestamp=%d -> %d",lastItemTime,item->timestamp));
            refClockTime = 0;
            refItemTime = 0;
            lastItemTime = 0;
            lastLookahead = 0;
            relativeItemTime = 0;
         }
         lastRequestId = item->requestId;
         lastDownloadId = item->downloadId;

         if (refClockTime == 0 || refItemTime == 0)
         {
            HandleQueueItem(item);
            STB_OSTaskDelay(1);
         }
         else
         {
            assert(clockTime >= refClockTime);
            assert(lastItemTime >= refItemTime);

            relativeClockTime = clockTime - refClockTime;
            relativeItemTime = lastItemTime - refItemTime;
            lastLookahead = relativeItemTime - relativeClockTime;

            DBG(("curr [%d] - ref [%d] = %d, "
                 "item [%d] - ref [%d] = %d, lla = %d",
                 clockTime, refClockTime, clockTime - refClockTime,
                 lastItemTime, refItemTime, lastItemTime - refItemTime, lastLookahead));
            if (lastLookahead > LOOK_AHEAD_DIFF)
            {
               DBG(("Flow control - pausing"));
               MHEG5QueueUpdateItem(item, 0);
               STB_OSTaskDelay((U16BIT)(lastLookahead-LOOK_AHEAD_DIFF));
            }
            else
            {
               HandleQueueItem(item);
               STB_OSTaskDelay(1);
            }
         }
      }
   }

   DBG(("TaskMain: Left main loop"));
   taskState = TASK_STOPPED;

   STB_OSDestroyTask(taskHandle);
}

static void HandleQueueItem(MHEG5QueueItem *item)
{
   U8BIT *data;
   U32BIT len;
   U32BIT processed;
   BOOLEAN last;

   /* Part of the item may have been processed already */
   data = item->data + item->offset;
   len = item->len - item->offset;
   last = item->last;

   DBG(("TaskMain: Sending %d/%d bytes (offset %d, last=%s)",
        len, item->len, item->offset, last ? "TRUE" : "FALSE"));

   /* Send data to external application */
   processed = DVB_MhegICStreamHandleData(data, len, last);
   if (processed == len)
   {
      /* The entire item was processed - release the item */
      MHEG5QueueReleaseItem(item);
   }
   else
   {
      /* Item still contains unprocessed data - update and keep it */
      MHEG5QueueUpdateItem(item, processed);
   }
}