glue_queue.c

Go to the documentation of this file.

/*******************************************************************************
 * Copyright © 2014 The DTVKit Open Software Foundation Ltd (www.dtvkit.org)
 * Copyright © 2011 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 "techtype.h"
#include "stb_os.h"
#include "glue_memory.h"
#include "glue_debug.h"
#include "glue_queue.h"

#define UP_MSGS_WHEN_Q_FULL
#ifndef NDEBUG
//#define MONITOR_STATS
#endif
#ifdef MONITOR_STATS
#define Q_STATS(s) s
#else
#define Q_STATS(s)
#endif
#define CRITICAL_Q_SIZE    (1 << 3)
#define HIGH_Q_SIZE        (1 << 3)
#define NORMAL_Q_SIZE      (1 << 8)
#define LOW_Q_SIZE         (1 << 4)

/* High priority section buffer exclusion space as percentage *
 * of total section queue size                                */
#define EXCLUSION_RATIO    (35)

typedef struct s_msg_queue
{
   U16BIT read_q_index;
   U16BIT write_q_index;
} S_MSG_QUEUE;

typedef struct s_q_notify
{
   struct s_q_notify *next;
   F_QueueNotify normal_rcvd_func;
   F_QueueNotify critical_rcvd_func;
   F_QueueNotify critical_done_func;
} S_Q_NOTIFY;

static void *vq_read_sema = 0;
static void *queue_mutex = 0;

static U16BIT queue_size[PRTY_INVALID] =
{ CRITICAL_Q_SIZE, HIGH_Q_SIZE, NORMAL_Q_SIZE, LOW_Q_SIZE };

static S_MSG_QUEUE *msg_queue[PRTY_INVALID];
static E_PRIORITY urgent_priority = PRTY_CRITICAL;
static S_Q_NOTIFY *vq_notify_head = NULL;
#ifdef TRACING
static int show_yeild_failure = 0;
#endif
#ifdef MONITOR_STATS
static int put_sema_total = 0;
static int get_sema_total = 0;
static int put_msg_total[PRTY_INVALID] = {0, 0, 0, 0, 0};
static int get_msg_total[PRTY_INVALID] = {0, 0, 0, 0, 0};
#if defined(STACK_DEBUGGING)
void STB_OSTaskPrint(void);
#endif
#endif

E_MhegErr VQ_Open( S_MhegConfig *cfg_params )
{
   int i;
   /* Check function parameters */
   if (queue_mutex != 0)
   {
      return MHERR_COMP_ALREADY_OPEN;
   }

   /* Create queue mutex */
   queue_mutex = STB_OSCreateMutex();
   if (queue_mutex == NULL)
   {
      TRACE(TERROR, ("failed to create mutex"));
      return MHERR_INTERNAL;
   }
   vq_read_sema = STB_OSCreateCountSemaphore(0);
   if (vq_read_sema == NULL)
   {
      TRACE(TERROR, ("failed to create semaphore"));
      return MHERR_INTERNAL;
   }
   /* Calculate num of high priority section buffers */
   for (i = 0; i != PRTY_INVALID; i++)
   {
      msg_queue[i] = SYS_Alloc( sizeof(S_MSG_QUEUE) + (queue_size[i] * sizeof(S_MhegMessage)));
      if (msg_queue[i] == NULL)
      {
         STB_OSDeleteMutex(queue_mutex);
         queue_mutex = 0;
         return MHERR_ALLOCATING_MEMORY;
      }
      msg_queue[i]->read_q_index = 0;
      msg_queue[i]->write_q_index = 0;
   }
   return MHERR_OK;
}

void VQ_Close(void)
{
   void *q_mtx;
   int i, nx;
   S_MhegMessage *p_elem;
   S_MSG_QUEUE *p_queue;
   FUNCTION_START(x)
   /* save queue_mutex and set to zero so that any other
    * task will already detect that queue is closing/closed */
   q_mtx = queue_mutex;
   queue_mutex = 0;
   if (q_mtx != 0)
   {
      /* now can close and release resources */
      i = PRTY_INVALID;

      STB_OSMutexLock(q_mtx);
      do
      {
         i--;
         /* Need to free data in queue items */
         p_queue = msg_queue[i];
         while (p_queue->read_q_index != p_queue->write_q_index)
         {
            nx = p_queue->read_q_index;
            p_elem = (S_MhegMessage *)(p_queue + 1);
            p_elem += nx;
            switch (p_elem->data_type)
            {
               default:
               case DT_NONE: case DT_VALUE:
                  /*nothing to do*/
                  break;

               case DT_ALLOC:
                  MHEG5freeMem((void *)p_elem->data.content.data );
                  break;

               case DT_CONTENT:
                  if (p_elem->data.content.destroy)
                  {
                     p_elem->data.content.destroy( p_elem->data.content.fs_handle );
                  }
                  break;

               case DT_D_CONTENT:
                  if (p_elem->data.content.destroy)
                  {
                     p_elem->data.content.destroy( &p_elem->data.content );
                  }
                  break;

               case DT_F_CONTENT:
                  p_elem->proc_msg_func( &(p_elem->data.content));
                  break;

               case DT_S_CONTENT:
                  STR_DataFree( p_elem->data.content.data, p_elem->data.content.size );
                  break;
            }
            nx++;
            if (nx == queue_size[i])
            {
               nx = 0;
            }
            p_queue->read_q_index = nx;
         }
         SYS_Free( p_queue );
      }
      while (i != 0);
      STB_OSMutexUnlock(q_mtx);
      STB_OSDeleteMutex(q_mtx);
      STB_OSDeleteSemaphore(vq_read_sema);
      vq_read_sema = 0;
   }
   FUNCTION_FINISH(x)
}

static U16BIT SizeFree( E_PRIORITY priority )
{
   U16BIT count;
   FUNCTION_START(x)
   assert( priority < PRTY_INVALID );
   //TRACE(TQUEUE,("%d c=%d,sz=%d,r=%d,w=%d",priority,count,queue_size[priority],
   //               msg_queue[priority]->read_q_index,msg_queue[priority]->write_q_index))
   if (msg_queue[priority]->read_q_index > msg_queue[priority]->write_q_index)
   {
      count = msg_queue[priority]->read_q_index - msg_queue[priority]->write_q_index;
   }
   else
   {
      count = queue_size[priority] + msg_queue[priority]->read_q_index - msg_queue[priority]->write_q_index;
   }
   FUNCTION_FINISH(x)
   return count;
}

E_MhegErr VQ_PutMsg( S_MhegMessage *pMsg, E_PRIORITY priority )
{
   E_MhegErr result = MHERR_OK;
   S_MhegMessage *p_elmnt;
   S_MSG_QUEUE *p_queue;
   S_Q_NOTIFY *vqn;
   U16BIT new_ndx;

   FUNCTION_START(x)
   assert( priority < PRTY_INVALID );
   if (queue_mutex == 0)
   {
      TRACE(TERROR, ("Queue not open"));
      result = MHERR_COMP_NOT_OPEN;
   }
   else
   {
      p_queue = msg_queue[priority];
      p_elmnt = (S_MhegMessage *)(p_queue + 1);

      STB_OSMutexLock(queue_mutex);

      new_ndx = p_queue->write_q_index;
      p_elmnt += new_ndx;
      new_ndx++;
      if (new_ndx == queue_size[priority])
      {
         new_ndx = 0;
      }
      if (new_ndx == p_queue->read_q_index)
      {
      #ifdef MONITOR_STATS
         int i, p = 0, g = 0;
         for (i = 0; i != PRTY_INVALID; i++)
         {
            p += put_msg_total[i];
            g += get_msg_total[i];
         }
         TRACE(TQUEUE, ("QQQ FULL sema=(%d,%d) qp=(%d,%d) all=(%d,%d) QQQ FULL",
                        put_sema_total, get_sema_total, put_msg_total[priority], get_msg_total[priority], p, g))
      #else
         TRACE(TWARN | TQUEUE, ("VGR QUEUE(%d) is FULL", priority))
      #endif
      #ifdef TRACING
         show_yeild_failure = 1;
      #endif
         result = MHERR_QUEUE_FULL;
      }
      else
      {
         if (priority == PRTY_CRITICAL)
         {
            vqn = vq_notify_head;
            while (vqn != NULL)
            {
               vqn->critical_rcvd_func();
               vqn = vqn->next;
            }
         }
         else if (priority != PRTY_LOW)
         {
            if (SizeFree(priority) < (queue_size[priority] >> 2))
            {
               urgent_priority = priority;
            }
            vqn = vq_notify_head;
            while (vqn != NULL)
            {
               vqn->normal_rcvd_func();
               vqn = vqn->next;
            }
         }
         memcpy( p_elmnt, pMsg, sizeof(S_MhegMessage));
         p_queue->write_q_index = new_ndx;
         Q_STATS(put_msg_total[priority]++; )
         Q_STATS(put_sema_total++; )
         STB_OSSemaphoreSignal(vq_read_sema);
      }
      STB_OSMutexUnlock(queue_mutex);
   }
   FUNCTION_FINISH(x)
   return result;
}

E_MhegErr VQ_GetMsg( S_MhegMessage *pMsg )
{
   E_MhegErr result = MHERR_INTERNAL;
   int i;
   S_MhegMessage *p_elem;
   S_MSG_QUEUE *p_queue;
   S_Q_NOTIFY *vqn;
   U16BIT nxt_ndx;

#ifdef MONITOR_STATS
   U32BIT starting;
   if (put_sema_total == get_sema_total)
   {
      starting = STB_OSGetClockMilliseconds();
   }
   else
   {
      starting = 0;
   }
#endif
   FUNCTION_START(x)
   STB_OSSemaphoreWait(vq_read_sema);
   Q_STATS(get_sema_total++; )
#ifdef MONITOR_STATS
   if (starting != 0)
   {
      TRACE(TQUEUE, ("QQQ TTT sema=(%d,%d) time=%d TTT QQQ", put_sema_total, get_sema_total, STB_OSGetClockMilliseconds() - starting))
   }
#endif

   STB_OSMutexLock(queue_mutex);
   if (urgent_priority != PRTY_CRITICAL &&
       msg_queue[PRTY_CRITICAL]->read_q_index != msg_queue[PRTY_CRITICAL]->write_q_index)
   {
      i = PRTY_CRITICAL;
   }
   else
   {
      i = urgent_priority;
   }
   for (; i != PRTY_INVALID; i++)
   {
      p_queue = msg_queue[i];
      if (p_queue->read_q_index != p_queue->write_q_index)
      {
         nxt_ndx = p_queue->read_q_index;
         p_elem = (S_MhegMessage *)(p_queue + 1);
         p_elem += nxt_ndx;
         nxt_ndx++;
         if (nxt_ndx == queue_size[i])
         {
            nxt_ndx = 0;
         }
         p_queue->read_q_index = nxt_ndx;
         memcpy( pMsg, p_elem, sizeof(S_MhegMessage));
         if (i == 0)
         {
            vqn = vq_notify_head;
            while (vqn != NULL)
            {
               vqn->critical_done_func();
               vqn = vqn->next;
            }
         }
         result = MHERR_OK;
         Q_STATS(get_msg_total[i]++; )
         break;
      }
   }
   if (urgent_priority != PRTY_CRITICAL &&
       SizeFree(urgent_priority) > ((queue_size[urgent_priority] * 3) >> 2))
   {
      urgent_priority = PRTY_CRITICAL;
   }
#ifdef MONITOR_STATS
   {
      int p = 0, g = 0;
      for (i = 0; i != PRTY_INVALID; i++)
      {
         p += put_msg_total[i];
         g += get_msg_total[i];
      }
      if (put_sema_total - get_sema_total != p - g)
      {
         TRACE(TQUEUE, ("QQQ sema=(%d,%d) all=(%d,%d) QQQ", put_sema_total, get_sema_total, p, g))
      }
   }
#endif
   STB_OSMutexUnlock(queue_mutex);
   FUNCTION_FINISH(x)
   return result;
}

U16BIT VQ_GetSizeFree( E_PRIORITY priority )
{
   U16BIT count;
   FUNCTION_START(x)
   assert( priority < PRTY_INVALID );
   if (queue_mutex == 0)
   {
      count = 0;
   }
   else
   {
      STB_OSMutexLock(queue_mutex);
      count = SizeFree(priority);
      STB_OSMutexUnlock(queue_mutex);
   }
   FUNCTION_FINISH(x)
   return count;
}

BOOLEAN VQ_EventNeedsProcessing( void )
{
   U16BIT priority;
   BOOLEAN needed = FALSE;
   FUNCTION_START(x)
   if (queue_mutex != NULL)
   {
      STB_OSMutexLock(queue_mutex);
      if (msg_queue[PRTY_CRITICAL]->read_q_index != msg_queue[PRTY_CRITICAL]->write_q_index)
      {
         /* have critical message - keypress stop/start or something */
         TRACE(TQUEUE, ("Critical MESSAGE"))
         needed = TRUE;
      }
      else
      {
         for (priority = PRTY_NORMAL; priority != PRTY_INVALID; priority++)
         {
            /* If space available on queue is less than 1/4 of total size,
             * then need to process q's */
            if (SizeFree(priority) < (queue_size[priority] >> 2))
            {
               urgent_priority = priority;
               TRACE(TQUEUE, ("Priority %d low on space", priority))
               needed = TRUE;
               break;
            }
         }
      }
      STB_OSMutexUnlock(queue_mutex);
   }
#ifdef TRACING
   if (!needed && show_yeild_failure)
   {
      for (priority = PRTY_NORMAL; priority != PRTY_INVALID; priority++)
      {
         TRACE(TQUEUE, ("p=%d, fsz=%d qsz=%d", priority, SizeFree(priority), queue_size[priority] >> 2));
      }
      show_yeild_failure = 0;
   }
#endif
   FUNCTION_FINISH(x)
   return needed;
}

void* VQ_RegisterNotify( F_QueueNotify normal_rcvd, F_QueueNotify critical_rcvd,
   F_QueueNotify critical_done )

{
   S_Q_NOTIFY *vqn;
   FUNCTION_START(x)
   vqn = SYS_Alloc( sizeof(S_Q_NOTIFY));
   if (vqn != NULL)
   {
      vqn->critical_rcvd_func = critical_rcvd;
      vqn->normal_rcvd_func = normal_rcvd;
      vqn->critical_done_func = critical_done;
      vqn->next = vq_notify_head;
      vq_notify_head = vqn;
   }
   FUNCTION_FINISH(x)
   return vqn;
}

void VQ_UnRegisterNotify( void *qn )
{
   S_Q_NOTIFY **ppvqn;
   FUNCTION_START(x)
   ppvqn = &vq_notify_head;
   while (*ppvqn)
   {
      if (*ppvqn == qn)
      {
         *ppvqn = (*ppvqn)->next;
         SYS_Free( qn );
         break;
      }
      ppvqn = &((*ppvqn)->next);
   }
   FUNCTION_FINISH(x)
}