rmpfs.c

Sample code for use on smp 863x processor.

/*****************************************
 Copyright 2001-2007
 Sigma Designs, Inc. All Rights Reserved
 Proprietary and Confidential
 *****************************************/
/**
  @file   rmpfs.c
  @brief  

  The pfs (Play For Sure) functions are used to start the
  bufferisation of files before the user asks for the playback.  This
  way when the user asks for a given url to be played, the first
  kilobytes of the files are already stored in memory and they can be
  demuxed and play without delay. The caller of these functions has to
  know which urls are likely to be played next so that they are
  already buffered when the user requests to play them.

  ** THIS IS NOT A REAL LIBRARY, since there's no 'handle', therefore only one
  instance can be used.

  How to use this lib:

  **************** Beginning of pseudo code ****************

  allocate a memory area to use by the prefetch function;
  fill up the pfs_profile struct;
  rmpfs_open(...);
  allocate a slot for each URL with rmpfs_allocate_slot_to_url(...);
  if (threaded) {
     LaunchThread(rmpfs_prefetch_slot(...));
     LaunchThread(rmpfs_prefetch_slot(...));
     ...
  }
  else {
     rmpfs_prefetch_slot(...);
     rmpfs_prefetch_slot(...);
     ...
  }
  play a given url;

  close the slots if necessary
  rmpfs_close_slot(...);
  rmpfs_close_slot(...);
  ...

  close the library before exiting the application
  rmpfs_close(...);

  ******************* End of pseudo code *******************

  **************************
  NOTE:
  **************************

  -Any seek call will interrupt prefetching.
  -The library works with or without threads.
  -This library is for 'reading' streams only, it cannot write to files.



  @author Sebastian Frias Feltrer
  @date   2007-01-15


  reported issues:
  
  bugct 2208: original feature request
  bugct 2719: allow prefetching to be interrupted by rmpfs_close_slot; 
              robustness
  bugct 2844: issue with seeking, seeking inside the cache is now possible;
              forbid multiple open request linked to the same url, only the first will pass thru cache.
  bugct 2885: do not close file handle on prefetch error, improved handling
              remove unnecessary critical sections
              reset stopPrefetch variable (might be linked to bugct 3029)


*/




#include <unistd.h>

#include "sample_os.h"

#include "rmpfs.h"
#include "rminputstream.h"

#include "rmupnp/rmlibwmdrmnd/include/ms_cardea.h"

#ifdef WITH_THREADS
#include "../rmlibcw/include/rmsemaphores.h"
#endif

/* #### Begin CARDEA code #### */
#include "rmupnp/rmlibwmdrmnd/include/ms_cardea.h"
/* #### End CARDEA code #### */



#if 1
#define LOCALDBG ENABLE
#else
#define LOCALDBG DISABLE
#endif

#if 0
#define READDBG ENABLE
#else
#define READDBG DISABLE
#endif

#if 0
#define SEEKDBG ENABLE
#else
#define SEEKDBG DISABLE
#endif


/*
  if KEEP_CACHE_AND_FILE_IN_SYNC is set to 1 then the file pointer will be 
  always synced with the cache pos;
  else, the file pointer will be synced with the cache pos only when a
  switch from cache to file is required.

  case 1) sync enabled:

  Everything works as usual until a seek call is performed. Prefetching will be
  interrupted to honor the seek. The file and pfs pointers will be set to the
  seeking destination.
  After this, all reads inside the cache will update the file pointer to match
  that of pfs.

  case 2) sync disabled:

  Everything works as usual until a seek call is performed. Prefetching will be
  interrupted to honor the seek. The file and pfs pointers will be set to the
  seeking destination.
  After this, if a seek to a destination inside the cache is performed only the
  pfs pointer will be updated (thus the non-synced nature of this mode). The file
  and pfs pointers will be updated only when a switch from cache to file is needed.

  case 1) is default although there are no differences from the functionality point
  of view.
*/

#define KEEP_CACHE_AND_FILE_IN_SYNC 1



#define PREFETCH_CHUNK_SIZE (4 * 1024)



#ifdef WITH_THREADS

#define ENTER_CS(x) do {			\
		RMEnterCriticalSection(x);	\
	} while (0);

#define LEAVE_CS(x) do {			\
		RMLeaveCriticalSection(x);	\
	} while (0);

#define SEMAPHORE_P(x) do {			\
		RMReleaseSemaphore(x, 1);	\
	} while (0);

#define SEMAPHORE_V(x) do {			\
		RMWaitForSemaphore(x);		\
	} while (0);
#else
#define ENTER_CS(x) do { } while(0);
#define LEAVE_CS(x) do { } while(0);
#define SEMAPHORE_P(x) do { } while(0);
#define SEMAPHORE_V(x) do { } while(0);
#endif

//FILE *saveFile;


static struct pfs_handle {
	struct pfs_profile *pfsProfile;
	RMuint32 url_prefetch_size;
	RMfile fileHandleTable[MAX_NUMBER_OF_CACHEABLE_URL];
	RMascii fileNameTable[MAX_NUMBER_OF_CACHEABLE_URL][256];
	RMbool urlAllocated[MAX_NUMBER_OF_CACHEABLE_URL];
	RMbool prefetchDone[MAX_NUMBER_OF_CACHEABLE_URL];
	RMbool stopPrefetch[MAX_NUMBER_OF_CACHEABLE_URL];

	/*
	  the slot in being used thru RMFile* calls by an application
	*/
	RMbool inUse[MAX_NUMBER_OF_CACHEABLE_URL];                
	
	/*
	  force all reads to be in sync with the original file handle
	*/
	RMbool forceSyncReads[MAX_NUMBER_OF_CACHEABLE_URL];

	RMuint32 prefetchedSize[MAX_NUMBER_OF_CACHEABLE_URL];

#ifdef WITH_THREADS
	RMsemaphore pfsOpenSemaphore[MAX_NUMBER_OF_CACHEABLE_URL];
	RMsemaphore pfsReadSemaphore[MAX_NUMBER_OF_CACHEABLE_URL];
	RMcriticalsection criticalSection[MAX_NUMBER_OF_CACHEABLE_URL];
#endif



} pfsHandle;




/* internal functions */

struct _pfs_cookie {
	RMuint32 slot;
	RMint64 position;
	
};


static RMint32 pfs_read(void *cookie, RMuint8 *buffer, RMuint32 size)
{
	struct _pfs_cookie *pfsCookie = (struct _pfs_cookie *)cookie;
	RMstatus status;
	RMint64 actualFilePos;

	RMint32 bytesLeft = size;
	RMint32 cache_size;
	RMint32 read_from_cache;
	RMint32 read_from_file = 0;
	RMint32 offset = 0;
	RMuint32 count = 0;
	

	RMGetCurrentPositionOfFile(pfsHandle.fileHandleTable[pfsCookie->slot], &actualFilePos);

	RMDBGLOG((READDBG, ">>>>\n"));
	RMDBGLOG((READDBG, "pfs_read(slot:%lu, size:%lu, buffer:%p)\n", pfsCookie->slot, size, buffer));
	RMDBGLOG((READDBG, "pos %lld (actual %lld), range[%lld, %lld], prefetchedSize %lu/%lu prefetchDone %lu\n",
		  pfsCookie->position,
		  actualFilePos,
		  pfsCookie->position,
		  pfsCookie->position + size,
		  pfsHandle.prefetchedSize[pfsCookie->slot],
		  pfsHandle.url_prefetch_size,
		  (RMuint32)pfsHandle.prefetchDone[pfsCookie->slot]));

	while (bytesLeft > 0) {
		RMbool forceSyncReads = FALSE;

		RMDBGLOG((READDBG, "bytesLeft %lu, offset %lu, pos %lld\n",
			  bytesLeft,
			  offset,
			  pfsCookie->position));


		if ((pfsCookie->position % PREFETCH_CHUNK_SIZE) == 0) {
			RMDBGLOG((READDBG, "check semaphore\n"));

			if (!pfsHandle.prefetchDone[pfsCookie->slot]) {
				SEMAPHORE_V(pfsHandle.pfsReadSemaphore[pfsCookie->slot]);
			}
		}
		cache_size = pfsHandle.prefetchedSize[pfsCookie->slot];

		count = 0;
		read_from_cache = RMmin(PREFETCH_CHUNK_SIZE, bytesLeft);
		read_from_cache = RMmin(read_from_cache, cache_size - pfsCookie->position);
		
		forceSyncReads = pfsHandle.forceSyncReads[pfsCookie->slot];

#if 0
		if (forceSyncReads) {
			// all reads will be from the file, caching disabled
			RMDBGLOG((READDBG, "forcing sync reads\n"));
			read_from_cache = 0;
		}
#endif


		if (read_from_cache > 0) {
			// read from cache

			RMDBGLOG((READDBG, "A: read_from_cache %ld read_from_file %ld cache_size %lu\n",
				  read_from_cache,
				  read_from_file,
				  cache_size));

			RMMemcpy(buffer + offset, (pfsHandle.pfsProfile->prefetch_area + (pfsCookie->slot * pfsHandle.url_prefetch_size) + pfsCookie->position), read_from_cache);
			
			count = read_from_cache;
			RMDBGLOG((READDBG, "read_from_cache %lu bytes\n", count));

		}
		else if (pfsHandle.prefetchDone[pfsCookie->slot]) {
			// read from file
			RMuint32 bytesRead;
		
			read_from_file = bytesLeft;

			RMDBGLOG((READDBG, "B: read_from_cache %ld read_from_file %ld cache_size %lu\n",
				  read_from_cache,
				  read_from_file,
				  cache_size));
	
			RMGetCurrentPositionOfFile(pfsHandle.fileHandleTable[pfsCookie->slot], &actualFilePos);
			
			RMDBGLOG((READDBG, "read_from_file[%lu] cachePos %llu, actualPos %llu\n", pfsCookie->slot, pfsCookie->position, actualFilePos));
			
			if (actualFilePos != pfsCookie->position) {
				RMDBGLOG((ENABLE, "[%lu] resyncing cache (%llu) and file (%llu) pointers\n", pfsCookie->slot, pfsCookie->position, actualFilePos));
				RMSeekFile(pfsHandle.fileHandleTable[pfsCookie->slot], pfsCookie->position, RM_FILE_SEEK_START);
			}
			
			status = RMReadFile(pfsHandle.fileHandleTable[pfsCookie->slot], buffer + offset, read_from_file, &bytesRead);
			if (status == RM_ERRORENDOFFILE) {
				RMDBGLOG((ENABLE, "[%lu] EOF\n", pfsCookie->slot));
				//quit loop gracefully
				bytesLeft = 0;
			}
			else if (status != RM_OK) {
				RMDBGLOG((ENABLE, "error %s prefetch slot %lu, aborting\n", RMstatusToString(status), pfsCookie->slot));
				return -1;
			}
			
			count = bytesRead;
			RMDBGLOG((READDBG, "read_from_file %lu bytes\n", count));
		}
		else
			RMDBGLOG((ENABLE, "need to wait\n"));

		bytesLeft -= count;
		pfsCookie->position += count;
		offset += count;

		if (bytesLeft < 0)
			RMDBGLOG((ENABLE, "bytesLeft %ld  is negative!!\n", bytesLeft));


#if KEEP_CACHE_AND_FILE_IN_SYNC
		if (forceSyncReads) {
			// sync read pointers
			
			RMGetCurrentPositionOfFile(pfsHandle.fileHandleTable[pfsCookie->slot], &actualFilePos);
			
			if (actualFilePos != pfsCookie->position) {
				RMDBGLOG((READDBG, "[%lu] Forced: resyncing cache (%llu) and file (%llu) pointers\n", pfsCookie->slot, pfsCookie->position, actualFilePos));
				RMSeekFile(pfsHandle.fileHandleTable[pfsCookie->slot], pfsCookie->position, RM_FILE_SEEK_START);
			}
		}
#endif
	}

	//fwrite(buffer, offset, 1, saveFile);

	RMGetCurrentPositionOfFile(pfsHandle.fileHandleTable[pfsCookie->slot], &actualFilePos);

	RMDBGLOG((READDBG, "total bytes read %lu curretFilePos %lld currentCachePos %lld\n", offset, actualFilePos, pfsCookie->position));
	RMDBGLOG((READDBG, "<<<<\n"));
	if (offset < 0)
		RMDBGLOG((ENABLE, "\n\noffset is negative! = %ld\n", offset));

	return offset;

}

static RMint32 pfs_write(void *cookie, const RMuint8 *buffer, RMuint32 size)
{
#ifndef NDEBUG
	struct _pfs_cookie *pfsCookie = (struct _pfs_cookie *)cookie;
#endif
	RMDBGLOG((LOCALDBG, "pfs_write(slot:%lu) not handled!!\n", pfsCookie->slot));

	return -1;
}

static RMint32 pfs_seek(void *cookie, RMint64 *position, RMfileSeekPos whence)
{
	struct _pfs_cookie *pfsCookie = (struct _pfs_cookie *)cookie;
	RMstatus status;
	RMint64 currentPosition;
	RMint32 error = 0;

	ENTER_CS(pfsHandle.criticalSection[pfsCookie->slot]);

	currentPosition = pfsCookie->position;

	RMDBGLOG((SEEKDBG, "pfs_seek(slot:%lu, pos:%lld, from:%lu [%s])\n", 
		  pfsCookie->slot, 
		  *position, 
		  (RMuint32)whence,
		  (whence == RM_FILE_SEEK_START) ? "from start": (whence == RM_FILE_SEEK_CURRENT) ? "current":"from the end" ));


	// if forceSyncReads is enabled, always perform real seeks
	if (KEEP_CACHE_AND_FILE_IN_SYNC && pfsHandle.forceSyncReads[pfsCookie->slot])
		goto perform_real_seek;


	switch (whence) {
	case RM_FILE_SEEK_START:
		if (*position < 0) {
			RMDBGLOG((ENABLE, "trying to seek beyond start of file\n"));
			error = -1;
			goto exit;
		}
		
		if (*position < pfsHandle.prefetchedSize[pfsCookie->slot]) {
			// seek is within cached data
			currentPosition = *position;

			RMDBGLOG((SEEKDBG, "seek to %lld is inside cache [0;%lu]\n", currentPosition, pfsHandle.prefetchedSize[pfsCookie->slot]));

			goto exit;
		}
		// else perform a 'real' seek and stop prefetching

		break;
	case RM_FILE_SEEK_CURRENT:

		currentPosition += *position;

		if (currentPosition < 0) {
			RMDBGLOG((ENABLE, "trying to seek beyond start of file\n"));
			error = -1;
			goto exit;
		}

		if (currentPosition < pfsHandle.prefetchedSize[pfsCookie->slot]) {
			// seek is within cached data

			RMDBGLOG((SEEKDBG, "seek to %lld is inside cache [0;%lu]\n", currentPosition, pfsHandle.prefetchedSize[pfsCookie->slot]));
			
			goto exit;
		}
		// else perform a 'real' seek and stop prefetching

		break;
	case RM_FILE_SEEK_END:
		if (*position > 0) {
			RMDBGLOG((ENABLE, "trying to seek beyond end of file\n"));
			error = -1;
			goto exit;
		}

		/* 
		   in order to handle this case correctly we'd need to know the length of the file,
		   we might not be able to do that in all cases, therefore we forward the request 
		   straight to the 'real' seek
		*/

		break;
	};

	if (!pfsHandle.prefetchDone[pfsCookie->slot]) 
		RMDBGLOG((ENABLE, "disabling prefetch for slot %lu (%lu bytes prefetched) since a real seek needs to be performed\n", pfsCookie->slot, pfsHandle.prefetchedSize[pfsCookie->slot]));