stmr_queue.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 <string.h>

#include "stb_os.h"
#include "glue_memory.h"
#include "glue_debug.h"

#include "mh5profile.h"
#include "stmr_queue.h"
#include "stmr_common.h"
#include "stmr_task.h"
#include "stmr_dl.h"
/*---constant definitions for this file--------------------------------------*/

#undef QDBG_FUNC
#define QDBG_CHECK
#undef QDBG_PRINT
#undef QDBG_ITEMS

#define QDBG_ALLOC(x)
#define QDBG_FREE(x)

#ifdef QDBG_FUNC
#define QDBG_FUNC_START(x) TRACE(TICS, (">> %s\n", #x))
#define QDBG_FUNC_END(x) TRACE(TICS, ("<< %s\n", #x))
#else
#define QDBG_FUNC_START(x)
#define QDBG_FUNC_END(x)
#endif

#ifdef QDBG_CHECK
#define QUEUE_CHECK() QueueCheck()
#else
#define QUEUE_CHECK()
#endif  /* QDBG_CHECK */

#ifdef QDBG_PRINT
#define QUEUE_PRINT() QueuePrint()
#else
#define QUEUE_PRINT()
#endif  /* QDBG_PRINT */

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


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

#define MIN_THRESHOLD_SIZE 1048576

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

/*---global function definitions---------------------------------------------*/

void SendSignalToTask(void);
#ifdef TO_REMOVE
BOOLEAN IsDataAvailableForStreaming(void);
#endif  /* TO_REMOVE */

static void* QueueAlloc(U32BIT size);
static void QueueFree(void *block);
#ifdef QDBG_CHECK
static void QueueCheck(void);
#endif
#ifdef QDBG_PRINT
static void QueuePrint(void);
#endif

static U32BIT QueueReleaseRequestItems(MHEG5QueueItem **head,
   MHEG5QueueItem **tail,
   U32BIT requestId);
static void QueueReleaseAllItems(MHEG5QueueItem **head, MHEG5QueueItem **tail);

/*---local (static) variable declarations for this file----------------------*/
static U32BIT queueAlign;
static U32BIT queueAlignMask;
static U8BIT *queueMemPool;
static U32BIT queueMemPoolSize;
static U32BIT queueMinBlockSize;
static U32BIT queueLastOffset;
static U32BIT queueThresholdSize;

static MHEG5QueueItem *queueItemHead = NULL;
static MHEG5QueueItem *queueItemTail = NULL;

static void *queueMutex = NULL;
static BOOLEAN queueStreamingEnabled = FALSE;
static U32BIT queueStreamingRequestId = 0;
static MHEG5QueueItem queueInvalidItem;
static U32BIT queueBufferedBytes;

static void (*insertCallback)(U32BIT downloadId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last) = 0;
static void (*releaseCallback)(U32BIT downloadId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last) = 0;


E_MhegErr MHEG5QueueInit(U8BIT *buffer, U32BIT bufferSize)
{
   E_MhegErr result;
   U32BIT size;

   queueMutex = STB_OSCreateMutex();
   if (queueMutex == NULL)
   {
      TRACE(TERROR, ("Streaming queue mutex failure"));
      result = MHERR_SYSTEM_RESOURCE_MUTEX;
   }
   else
   {
      queueItemHead = NULL;
      queueItemTail = NULL;

      queueBufferedBytes = 0;

      /* Queue manager is a mini memory manager. It uses the provided buffer
       * as the memory area to allocate memory from. It's a general purpose
       * memory manager, used for items as well as data blocks.
       *
       * Used blocks hold the size at the start of the block. The block
       * size is always a multiple of 4, so the two LSBs are "unused" for the
       * size. So the memory manager uses them to find out whether the
       * previous block is used (bit 0) and whether the current block is used
       * (bit 1).
       *
       * Unused blocks also contain the size at the end of the block.
       *
       * Used block:    | size | this used | prev used | data         |
       * Unused block:  | size | this used | prev used | empty | size |
       *
       * This structure allows unused blocks to be merged, while keeping
       * overhead to a minimum.
       *
       * Initial setting is two blocks:
       * 1. One big block at the start with "this used" false and "prev used"
       *    true (so that it is never merged with the previous block, which
       *    doesn't exist)
       * 2. One small block at the end with "this used" true and "prev used"
       *    false (so that it is never merged with the large block).
       *
       * This allows allocation / free functions to be generic without worrying
       * about first / last block.
       */

      queueAlign = sizeof(U32BIT) * 2;
      queueAlignMask = ~(queueAlign - 1);
      queueMemPool = buffer;
      queueMemPoolSize = bufferSize & queueAlignMask;
      queueMinBlockSize = 2 * sizeof(U32BIT);
      queueLastOffset = 0;
      if (bufferSize < MIN_THRESHOLD_SIZE)
      {
         queueThresholdSize = bufferSize;
      }
      else
      {
         queueThresholdSize = MIN_THRESHOLD_SIZE + ((bufferSize-MIN_THRESHOLD_SIZE)/2);
      }
      result = MHERR_OK;
      if (queueMemPoolSize >= 2 * queueMinBlockSize)
      {
         /* Create two blocks - one "real" to use for allocations, and one
          * "fake" as a sentinel at the end of the buffer
          */
         size = queueMemPoolSize - queueMinBlockSize;
         assert((size & 0x3) == 0);

         /* The first block is unused, so the size is copied to the end of the
          * block
          */
         *((U32BIT *)(queueMemPool + size - sizeof(U32BIT))) = size;

         /* The "previous" block is marked as "used". This stops any attempts to
          * merge the block with the non-existent previous block. Current block
          * is not used.
          */
         size |= 0x1;
         *((U32BIT *)queueMemPool) = size;

         /* We mark the sentinel block as "used" to stop any attempts to merge
          * it with previous blocks. This allows the allocation functions never
          * to worry about first/last blocks. The previous block is marked as
          * "unused".
          */
         *((U32BIT *)(queueMemPool + (size & ~0x3))) = queueMinBlockSize | 0x2;
      }
      else
      {
         /* Buffer too small */
         DBG(("QueueInit: Buffer too small"));
         queueMemPool = NULL;
         queueMemPoolSize = 0;

      #ifdef ERROR_ON_SMALL_ICS_BUFFER
         STB_OSDeleteMutex( queueMutex );
         queueMutex = NULL;
         result = MHERR_BAD_ICS_BUFFER_SIZE;
      #endif
      }
   }
   return result;
}

MHEG5QueueItem* MHEG5QueueAllocItem(U32BIT requestId, U32BIT len)
{
   MHEG5QueueItem *item;

   QDBG_FUNC_START(MHEG5QueueAllocItem);

   STB_OSMutexLock(queueMutex);

   item = QueueAlloc(sizeof *item);
   if (item != NULL)
   {
      item->data = QueueAlloc(len);
      if (item->data != NULL)
      {
         item->requestId = requestId;
         item->len = 0;
         item->offset = 0;
         item->referenceCount = 1;
      }
      else
      {
         QueueFree(item);
         item = NULL;
      }
   }

   STB_OSMutexUnlock(queueMutex);

   QDBG_FUNC_END(MHEG5QueueAllocItem);

   return item;
}

void MHEG5QueueInsertItem(MHEG5QueueItem *item)
{
   QDBG_FUNC_START(MHEG5QueueInsertItem);

   STB_OSMutexLock(queueMutex);

   item->next = NULL;
   if (queueItemHead == NULL)
   {
      assert(queueItemTail == NULL);
      queueItemHead = item;
      queueItemTail = item;
   }
   else
   {
      assert(queueItemTail != NULL);
      queueItemTail->next = item;
      queueItemTail = item;
   }

   queueBufferedBytes += item->len;
   DBG(("queueBufferedBytes = %d [1]", queueBufferedBytes));

    #ifdef QDBG_ITEMS
   {
      MHEG5QueueItem *item = queueItemHead;

      while (item != NULL)
      {
         DBG(("%p (%p,%d) -> ", item, item->data, item->len));
         item = item->next;
      }
      DBG(("nil"));

      DBG(("tail = %p", queueItemTail));
   }
    #endif

   STB_OSMutexUnlock(queueMutex);

   if (insertCallback != NULL)
   {
      /* Notify item insertion (queueMutex should be unlocked) */
      insertCallback(item->downloadId, item->base, item->position,
         item->len, item->last);
   }

   QDBG_FUNC_END(MHEG5QueueInsertItem);
}

MHEG5QueueItem* MHEG5QueueGetHeadItem(void)
{
   MHEG5QueueItem *item;

   QDBG_FUNC_START(MHEG5QueueGetHeadItem);

   STB_OSMutexLock(queueMutex);

   if (queueStreamingEnabled)
   {
      item = queueItemHead;
      if (item != NULL)
      {
         if (item->requestId == queueStreamingRequestId)
         {
            /* Item can be released */
            ++item->referenceCount;
         }
         else
         {
            /* Item belongs to another stream, pretend that
             * the queue is empty.
             */
            item = NULL;
         }
      }
   }
   else
   {
      /* Either streaming is not allowed, or there's not enough data
       * for streaming (or both).
       */
      item = &queueInvalidItem;
   }

   STB_OSMutexUnlock(queueMutex);

   QDBG_FUNC_END(MHEG5QueueGetHeadItem);

   return item;
}

BOOLEAN MHEG5QueueIsValidItem(MHEG5QueueItem *item)
{
   BOOLEAN valid = TRUE;

   if (item == &queueInvalidItem)
   {
      valid = FALSE;
   }

   return valid;
}

void MHEG5QueueUpdateItem(MHEG5QueueItem *item, U32BIT processed)
{
   BOOLEAN call;
   U32BIT requestId = 0;
   U64BIT base;
   U64BIT position;
   U32BIT len = 0;
   BOOLEAN last = FALSE;

   QDBG_FUNC_START(MHEG5QueueUpdateItem);

   STB_OSMutexLock(queueMutex);

   call = FALSE;
   --item->referenceCount;

   /* If the item is in the linked list, it's at the head of the list */
   if (item == queueItemHead)
   {
      item->offset += processed;
   }
   else
   {
      /* Updating causes the item to be released */
      call = TRUE;

      requestId = item->requestId;
      base = item->base;
      position = item->position;
      len = item->len;
      last = item->last;

      /* The item is not in the queue, all references are lost and it
       * must be freed.
       */
      QueueFree(item->data);
      QueueFree(item);
   }

   #ifdef QDBG_ITEMS
   {
      MHEG5QueueItem *item = queueItemHead;

      while (item != NULL)
      {
         DBG(("%p (%p,%d) -> ", item, item->data, item->len));
         item = item->next;
      }
      DBG(("nil"));

      DBG(("tail = %p", queueItemTail));
   }
   #endif

   STB_OSMutexUnlock(queueMutex);

   if (call && releaseCallback != NULL)
   {
      releaseCallback(requestId, base, position, len, last);
   }

   QDBG_FUNC_END(MHEG5QueueUpdateItem);
}

void MHEG5QueueReleaseItem(MHEG5QueueItem *item)
{
   U32BIT requestId;
   U64BIT base;
   U64BIT position;
   U32BIT len;
   BOOLEAN last, resume;

   QDBG_FUNC_START(MHEG5QueueReleaseItem);

   STB_OSMutexLock(queueMutex);

   /* Item information is passed to the callback */
   requestId = item->requestId;
   base = item->base;
   position = item->position;
   len = item->len;
   last = item->last;
   resume = FALSE;

   --item->referenceCount;

   /* If the item is in the linked list, it's at the head of the list */
   if (item == queueItemHead)
   {
      /* Remove item from item list */
      queueItemHead = queueItemHead->next;
      if (item == queueItemTail)
      {
         assert(queueItemHead == NULL);
         queueItemTail = NULL;
      }

      queueBufferedBytes -= item->len;
      DBG(("queueBufferedBytes = %d [2]", queueBufferedBytes));

      if (queueBufferedBytes < queueThresholdSize)
      {
         resume = TRUE;
      }
      /* The item is freed only if it's not currently used by the task */
      --item->referenceCount;
      if (item->referenceCount == 0)
      {
         QueueFree(item->data);
         QueueFree(item);
      }
   }
   else
   {
        PDBG(("Qbytes=%u, item->referenceCount=%d, item->len=%u", queueBufferedBytes, item->referenceCount, item->len));
      if (item->referenceCount <= 1)
      {
         assert(item->referenceCount == 0);
         QueueFree(item->data);
         QueueFree(item);
      }
   }

    #ifdef QDBG_ITEMS
   {
      MHEG5QueueItem *item = queueItemHead;

      while (item != NULL)
      {
         DBG(("%p (%p,%d) -> ", item, item->data, item->len));
         item = item->next;
      }
      DBG(("nil"));

      DBG(("tail = %p", queueItemTail));
   }
    #endif

   STB_OSMutexUnlock(queueMutex);

   if (releaseCallback != NULL)
   {
      /* Notify item released (queueMutex should be unlocked) */
      releaseCallback(requestId, base, position, len, last);
   }
   if (resume)
   {
      MHEG5ResumeDownload();
   }

   QDBG_FUNC_END(MHEG5QueueReleaseItem);
}

U32BIT MHEG5QueueGetBufferedBytes(void)
{
   QDBG_FUNC_START(MHEG5QueueGetBufferedBytes);

   QDBG_FUNC_END(MHEG5QueueGetBufferedBytes);

   return queueBufferedBytes;
}

void MHEG5QueueEnableStreaming(U32BIT requestId)
{
   BOOLEAN enabled;

   QDBG_FUNC_START(MHEG5QueueEnableStreaming);

   STB_OSMutexLock(queueMutex);
   PDBG(("QueueEnableStreaming"));
   enabled = queueStreamingEnabled;
   queueStreamingEnabled = TRUE;
   queueStreamingRequestId = requestId;
   if (!enabled)
   {
      /* Also signal to the task that reading was enabled */
      MHEG5StreamerSendSignalToTask();
   }
   STB_OSMutexUnlock(queueMutex);

   QDBG_FUNC_END(MHEG5QueueEnableStreaming);
}

void MHEG5QueueDisableStreaming(void)
{
   QDBG_FUNC_START(MHEG5QueueDisableStreaming);

   STB_OSMutexLock(queueMutex);
   PDBG(("QueueDisableStreaming"));
   queueStreamingEnabled = FALSE;
   queueStreamingRequestId = 0;
   STB_OSMutexUnlock(queueMutex);

   QDBG_FUNC_END(MHEG5QueueDisableStreaming);
}

BOOLEAN MHEG5QueueIsStreamingEnabled(void)
{
   return queueStreamingEnabled;
}

void MHEG5QueueReleaseRequestItems(U32BIT requestId)
{
   BOOLEAN resume;
   QDBG_FUNC_START(MHEG5QueueReleaseRequestItems);

   STB_OSMutexLock(queueMutex);

   queueBufferedBytes -= QueueReleaseRequestItems(&queueItemHead,
         &queueItemTail,
         requestId);

   resume = (queueBufferedBytes < queueThresholdSize)? TRUE : FALSE;

   #ifdef QDBG_ITEMS
   {
      MHEG5QueueItem *item = queueItemHead;

      while (item != NULL)
      {
         DBG(("%p (%p,%d) -> ", item, item->data, item->len));
         item = item->next;
      }
      DBG(("nil"));

      DBG(("tail = %p", queueItemTail));
   }
   #endif

   STB_OSMutexUnlock(queueMutex);

   if (resume)
   {
      MHEG5ResumeDownload();
   }

   QDBG_FUNC_END(MHEG5QueueReleaseRequestItems);
}

void MHEG5QueueReleaseItems(U32BIT requestId, U64BIT marker)
{
   MHEG5QueueItem **pItem, *item;
   U64BIT pos;
   BOOLEAN resume;
   QDBG_FUNC_START(MHEG5QueueReleaseItems);

   STB_OSMutexLock(queueMutex);
   pItem = &queueItemHead;
   while (*pItem != NULL)
   {
      item = *pItem;
      if (item->requestId != requestId || ULL_Compare(item->position,marker) > 0)
      {
         pItem = &item->next;
      }
      else
      {
         pos = item->position;
         ULL_Add32(pos,item->len);
         if (ULL_Compare(pos,marker) <= 0)
         {
            *pItem = item->next;
            queueBufferedBytes -= item->len;
            --item->referenceCount;
            if (item->referenceCount == 0)
            {
                QueueFree(item->data);
                QueueFree(item);
            }
         }
         else
         {
            pos = marker;
            ULL_Sub(pos,item->position);
            item->offset = ULL_Get32(pos);
            pItem = &item->next;
         }
      }
   }
   resume = (queueBufferedBytes < queueThresholdSize)? TRUE : FALSE;

   /* Find the tail of the list */
   item = queueItemHead;
   if (item == NULL)
   {
      queueItemTail = NULL;
   }
   else
   {
      while (item->next != NULL)
      {
         item = item->next;
      }
      queueItemTail = item;
   }
   STB_OSMutexUnlock(queueMutex);

   if (resume)
   {
      MHEG5ResumeDownload();
   }
   QDBG_FUNC_END(MHEG5QueueReleaseItems);
}

void MHEG5QueueReleaseAllItems(void)
{
   QDBG_FUNC_START(MHEG5QueueReleaseAllItems);

   STB_OSMutexLock(queueMutex);

   QueueReleaseAllItems(&queueItemHead, &queueItemTail);
   queueBufferedBytes = 0;

   DBG(("queueBufferedBytes = 0 [4]"));

   STB_OSMutexUnlock(queueMutex);

   QDBG_FUNC_END(MHEG5QueueReleaseAllItems);
}

void MHEG5QueueRegisterInsertCallback(void (*callback)(U32BIT downloadId,
      U64BIT base,
      U64BIT position,
      U32BIT len,
      BOOLEAN last))
{
   insertCallback = callback;
}

void MHEG5QueueRegisterReleaseCallback(void (*callback)(U32BIT requestId,
      U64BIT base,
      U64BIT position,
      U32BIT len,
      BOOLEAN last))
{
   releaseCallback = callback;
}

void MHEG5QueueTerm(void)
{
   void *mutex;

   STB_OSMutexLock(queueMutex);

   queueMemPool = NULL;
   queueMemPoolSize = 0;

   mutex = queueMutex;
   queueMutex = NULL;

   STB_OSMutexUnlock(mutex);
   STB_OSDeleteMutex(mutex);
}

static void* QueueAlloc(U32BIT size)
{
   U32BIT offset;
   U32BIT blockHeader;
   U32BIT blockSize;
   U32BIT requestSize;
   U32BIT newHeader;
   U8BIT *newHeaderPtr;
   void *block;

   QDBG_ALLOC((">> QueueAlloc(%d)\n", size));

   block = NULL;

   /* Make sure the size uses the alignment */
   requestSize = (size + sizeof(U32BIT) + queueAlign - 1) & queueAlignMask;
   QDBG_ALLOC(("size = %d, requestSize = %d\n", size, requestSize));

   QDBG_ALLOC(("queueLastOffset = %d\n", queueLastOffset));
   offset = queueLastOffset;
   while (block == NULL && offset < queueMemPoolSize)
   {
      blockHeader = *((U32BIT *)(queueMemPool + offset));
      blockSize = blockHeader & ~0x3;
      if ((blockHeader & 0x2) || (blockSize < requestSize))
      {
         /* Current block is used, or not large enough to satisfy the
          * request. Move on to the next block.
          */
         QDBG_ALLOC(("Nothing at %d, trying next offset %d\n",
                     offset, offset + blockSize));
         offset += blockSize;
         if (offset == queueMemPoolSize)
         {
            /* Back to start of the buffer */
            offset = 0;
         }
         if (offset == queueLastOffset)
         {
            /* Back to where we started, all blocks are used */
            break;
         }
      }
      else
      {
         /* Found a suitable block - take only what we need */
         QDBG_ALLOC(("Found a block - blockSize = %d\n", blockSize));
         if (blockSize - requestSize >= queueMinBlockSize)
         {
            /* Split the block into two */
            QDBG_ALLOC(("Need to split the block\n"));
            newHeader = requestSize;
            /* Copy "previous used" flag from original header */
            newHeader |= blockHeader & 0x1;
            /* Current block is marked as "used" */
            newHeader |= 0x2;
            /* Write the header into the buffer */
            *((U32BIT *)(queueMemPool + offset)) = newHeader;

            /* Next block is requestSize bytes away */
            newHeaderPtr = queueMemPool + offset + requestSize;
            newHeader = blockSize - requestSize;
            /* Update block size at the end of the unused block */
            *((U32BIT *)(newHeaderPtr + newHeader - sizeof(U32BIT))) = newHeader;
            /* "previous used" is 1, we've just created it. "current used"
             * is 0 because we started with an unused block. */
            newHeader |= 0x1;
            /* Write the header into the buffer */
            *((U32BIT *)newHeaderPtr) = newHeader;

            /* Note that the block after that doesn't have to change. It's
             * "previous used" flag is 0 (because the block was unused),
             * and because we've split the block into two, that's still the
             * case.
             */

            QUEUE_CHECK();
         }
         else
         {
            /* Use the entire block */
            QDBG_ALLOC(("Use entire block\n"));
            requestSize = blockSize;
            newHeader = requestSize;
            /* Copy "previous used" flag from original header */
            newHeader |= blockHeader & 0x1;
            /* Current block is marked as "used" */
            newHeader |= 0x2;
            /* Write the header into the buffer */
            *((U32BIT *)(queueMemPool + offset)) = newHeader;

            /* Next block's "previous used" flag must be updated to
             * reflect the fact that we've allocate a block
             */
            *((U32BIT *)(queueMemPool + offset + requestSize)) |= 0x1;

            QUEUE_CHECK();
         }

         block = queueMemPool + offset + sizeof(U32BIT);

         /* Look for the next block from the end of this block onwards */
         queueLastOffset = offset + requestSize;
      }
   }

   QDBG_ALLOC(("<< QueueAlloc() -> %p\n", block));

   return block;
}

static void QueueFree(void *block)
{
   U8BIT *blockHeaderPtr;
   U8BIT *nextBlockHeaderPtr;
   U8BIT *prevBlockHeaderPtr;
   U32BIT blockHeader;
   U32BIT blockOffset;
   U32BIT blockSize;
   U32BIT nextBlockSize;
   U32BIT prevBlockSize;
   U32BIT prevBlockHeader;

   QDBG_FREE((">> QueueFree(%p)\n", block));

   /* Freeing a block is done in stages. There are 4 possible scenarios
    * where previous/next block are used/unused. Instead of checking each
    * scenario separately, we just follow the following (slightly inefficient)
    * sequence:
    * 1. Free the block without any merging
    * 2. If the next block is unused, merge with next block.
    * 3. If the previous block is unused, merge with previous block.
    */

   blockOffset = (U8BIT *)block - sizeof(U32BIT) - queueMemPool;
   assert(blockOffset < queueMemPoolSize);

   blockHeaderPtr = queueMemPool + blockOffset;
   blockHeader = *((U32BIT *)blockHeaderPtr);
   blockSize = blockHeader & ~0x3;

   /* Make sure the block is marked as "used" */
   assert(blockHeader & 0x2);

   /* Mark current block as "unused" */
   blockHeader &= ~0x2;
   *((U32BIT *)blockHeaderPtr) = blockHeader;

   /* Copy block size to the end of the block */
   *((U32BIT *)(blockHeaderPtr + blockSize - sizeof(U32BIT))) = blockSize;

   /* Update next block */
   nextBlockHeaderPtr = blockHeaderPtr + blockSize;
   *((U32BIT *)nextBlockHeaderPtr) &= ~0x1;

   QUEUE_CHECK();

   if ((*((U32BIT *)nextBlockHeaderPtr) & 0x2) == 0)
   {
      /* Next block is unused, merge with next block */
      nextBlockSize = *((U32BIT *)nextBlockHeaderPtr) & ~0x3;

      /* Update current header */
      blockSize += nextBlockSize;
      blockHeader = blockSize | (blockHeader & 0x3);
      *((U32BIT *)blockHeaderPtr) = blockHeader;

      /* Copy next block size to the end of the block */
      *((U32BIT *)(blockHeaderPtr + blockSize - sizeof(U32BIT))) = blockSize;

      /* No need to update the block after that, as it already indicates
       * that its previous block is unused.
       */

      if (queueLastOffset == nextBlockHeaderPtr - queueMemPool)
      {
         /* Last offset is invalid, make it valid again */
         queueLastOffset = blockOffset;
      }
      QUEUE_CHECK();
   }

   if ((blockHeader & 0x1) == 0)
   {
      /* Previous block is unused, merge with current block */
      prevBlockSize = *((U32BIT *)(blockHeaderPtr - sizeof(U32BIT)));
      prevBlockHeaderPtr = blockHeaderPtr - prevBlockSize;
      prevBlockHeader = *((U32BIT *)prevBlockHeaderPtr);

      /* Update previous block header */
      prevBlockSize += blockSize;
      prevBlockHeader = prevBlockSize | (prevBlockHeader & 0x3);
      *((U32BIT *)prevBlockHeaderPtr) = prevBlockHeader;

      /* Copy previous block size to the end of the block */
      *((U32BIT *)(prevBlockHeaderPtr + prevBlockSize - sizeof(U32BIT))) = prevBlockSize;

      if (queueLastOffset == blockOffset)
      {
         /* Last offset is invalid, make it valid again */
         queueLastOffset = prevBlockHeaderPtr - queueMemPool;
      }
      QUEUE_CHECK();
   }

   QDBG_FREE(("<< QueueFree()\n"));
}

#ifdef QDBG_CHECK

static void QueueCheck(void)
{
   U32BIT offset;
   U32BIT size;
   U32BIT prevUsed, currUsed, realPrevUsed;
   U32BIT lastWord;
   BOOLEAN seenLastOffset;

   QUEUE_PRINT();

   offset = 0;
   realPrevUsed = 1;
   seenLastOffset = FALSE;

   while (offset < queueMemPoolSize)
   {
      if (queueLastOffset == offset)
      {
         seenLastOffset = TRUE;
      }
      size = *((U32BIT *)(queueMemPool + offset));
      prevUsed = size & 0x1;
      currUsed = (size >> 1) & 0x1;
      size &= ~0x3;
      assert(offset + size <= queueMemPoolSize);
      assert(prevUsed == realPrevUsed);
      realPrevUsed = currUsed;
      lastWord = *((U32BIT *)(queueMemPool + offset + size - sizeof(U32BIT)));
      assert(currUsed || lastWord == size);
      offset += size;
   }

   assert(seenLastOffset);
}

#endif  /* QDBG_CHECK */


#ifdef QDBG_PRINT

static void QueuePrint(void)
{
   U32BIT offset;
   U32BIT size;
   U32BIT prevUsed, currUsed;
   U32BIT lastWord;

   DBG((">> QueuePrint"));

   offset = 0;
   while (offset < queueMemPoolSize)
   {
      size = *((U32BIT *)(queueMemPool + offset));
      prevUsed = size & 0x1;
      currUsed = (size >> 1) & 0x1;
      size &= ~0x3;
      lastWord = *((U32BIT *)(queueMemPool + offset + size - sizeof(U32BIT)));
      DBG(("%p: offset=%d, size=%d, prevUsed=%d, currUsed=%d, lastWord=%d",
           queueMemPool + offset + sizeof(U32BIT),
           offset, size, prevUsed, currUsed, lastWord));
      offset += size;
   }

   DBG(("<< QueuePrint"));
}

#endif  /* QDBG_PRINT */


static U32BIT QueueReleaseRequestItems(MHEG5QueueItem **head,
   MHEG5QueueItem **tail,
   U32BIT requestId)
{
   MHEG5QueueItem **pItem, *item;
   U32BIT released;

   QDBG_FUNC_START(QueueReleaseRequestItems);

   released = 0;

   pItem = head;
   while (*pItem != NULL)
   {
      item = *pItem;
      if (item->requestId == requestId)
      {
         *pItem = item->next;

         released += item->len;

         /* The item is freed only if it's not currently used by the task */
         --item->referenceCount;
         if (item->referenceCount == 0)
         {
            QueueFree(item->data);
            QueueFree(item);
         }
      }
      else
      {
         pItem = &item->next;
      }
   }

   /* Find the tail of the list */
   *tail = *head;
   while (*tail != NULL && (*tail)->next != NULL)
   {
      *tail = (*tail)->next;
   }

   QDBG_FUNC_END(QueueReleaseRequestItems);

   return released;
}

static void QueueReleaseAllItems(MHEG5QueueItem **head, MHEG5QueueItem **tail)
{
   MHEG5QueueItem *item, *nextItem;

   QDBG_FUNC_START(QueueReleaseAllItems);

   item = *head;
   while (item != NULL)
   {
      nextItem = item->next;

      /* The item is freed only if it's not currently used by the task */
      --item->referenceCount;
      if (item->referenceCount == 0)
      {
         QueueFree(item->data);
         QueueFree(item);
      }

      item = nextItem;
   }

   *head = NULL;
   *tail = NULL;

   QDBG_FUNC_END(QueueReleaseAllItems);
}