* Completely rewrote the ChunkCache - the previous version had some issues with

regards to locking and seeking.
* Furthermore, we now not only cache 4 chunks, but chunk up to a certain
  memory size (MediaExtractor uses 1 MB for now).
* Since I still have occasional hickups, it looks like this wasn't the main
  cause for our audio problems. Still, this will reduce drive access
  considerably during play.


git-svn-id: file:///srv/svn/repos/haiku/haiku/trunk@34243 a95241bf-73f2-0310-859d-f6bbb57e9c96

headers/private/media/MediaExtractor.h

@@ -1,6 +1,8 @@
 /*
  * Copyright 2004-2007, Marcus Overhagen. All rights reserved.
  * Copyright 2008, Maurice Kalinowski. All rights reserved.
+ * Copyright 2009, Axel Dörfler, [email protected].
+ *
  * Distributed under the terms of the MIT License.
  */
 #ifndef _MEDIA_EXTRACTOR_H
@@ -14,7 +16,10 @@
 namespace BPrivate {
 namespace media {
 
+
 class ChunkCache;
+struct chunk_buffer;
+
 
 struct stream_info {
 	status_t		status;
@@ -23,9 +28,11 @@ struct stream_info {
 	const void*		infoBuffer;
 	size_t			infoBufferSize;
 	ChunkCache*		chunkCache;
+	chunk_buffer*	lastChunk;
 	media_format	encodedFormat;
 };
 
+
 class MediaExtractor {
 public:
 								MediaExtractor(BDataIO* source, int32 flags);
@@ -58,6 +65,7 @@ public:
 									media_codec_info* codecInfo);
 
 private:
+			void				_RecycleLastChunk(stream_info& info);
 	static	int32				_ExtractorEntry(void* arg);
 			void				_ExtractorThread();
 
@@ -66,7 +74,6 @@ private:
 
 			sem_id				fExtractorWaitSem;
 			thread_id			fExtractorThread;
-			volatile bool		fTerminateExtractor;
 
 			BDataIO*			fSource;
 			Reader*				fReader;

src/kits/media/ChunkCache.cpp

@@ -1,143 +1,144 @@
 /*
- * Copyright 2004, Marcus Overhagen. All rights reserved.
+ * Copyright 2009, Axel Dörfler, [email protected].
  * Distributed under the terms of the MIT License.
  */
 
 
 #include "ChunkCache.h"
 
-#include <string.h>
+#include <new>
+#include <stdlib.h>
 
-#include <Autolock.h>
+#include <Debug.h>
-
-#include "debug.h"
 
 
-ChunkCache::ChunkCache()
+chunk_buffer::chunk_buffer()
 	:
-	fLocker("media chunk cache locker")
+	buffer(NULL),
+	size(0),
+	capacity(0)
 {
-	// fEmptyChunkCount must be one less than the real chunk count,
+}
-	// because the buffer returned by GetNextChunk must be preserved
-	// until the next call of that function, and must not be overwritten.
 
-	fEmptyChunkCount = CHUNK_COUNT - 1;
-	fReadyChunkCount = 0;
-	fNeedsRefill = 1;
 
-	fGetWaitSem = create_sem(0, "media chunk cache sem");
+chunk_buffer::~chunk_buffer()
+{
+	free(buffer);
+}
 
-	fNextPut = &fChunkInfos[0];
-	fNextGet = &fChunkInfos[0];
 
-	for (int i = 0; i < CHUNK_COUNT; i++) {
+// #pragma mark -
-		fChunkInfos[i].next = i == CHUNK_COUNT - 1
+
-			? &fChunkInfos[0] : &fChunkInfos[i + 1];
+
-		fChunkInfos[i].buffer = NULL;
+ChunkCache::ChunkCache(sem_id waitSem, size_t maxBytes)
-		fChunkInfos[i].sizeUsed = 0;
+	:
-		fChunkInfos[i].sizeMax = 0;
+	BLocker("media chunk cache"),
-		fChunkInfos[i].status = B_ERROR;
+	fWaitSem(waitSem),
-	}
+	fMaxBytes(maxBytes),
+	fBytes(0)
+{
 }
 
 
 ChunkCache::~ChunkCache()
 {
-	delete_sem(fGetWaitSem);
+	while (chunk_buffer* chunk = fChunks.RemoveHead())
+		delete chunk;
 
-	for (int i = 0; i < CHUNK_COUNT; i++) {
+	while (chunk_buffer* chunk = fUnusedChunks.RemoveHead())
-		free(fChunkInfos[i].buffer);
+		delete chunk;
-	}
+
+	while (chunk_buffer* chunk = fInFlightChunks.RemoveHead())
+		delete chunk;
 }
 
 
 void
 ChunkCache::MakeEmpty()
 {
-	BAutolock _(fLocker);
+	ASSERT(IsLocked());
 
-	fEmptyChunkCount = CHUNK_COUNT - 1;
+	fUnusedChunks.MoveFrom(&fChunks);
-	fReadyChunkCount = 0;
+	fBytes = 0;
-	fNextPut = &fChunkInfos[0];
+
-	fNextGet = &fChunkInfos[0];
+	release_sem(fWaitSem);
-	atomic_or(&fNeedsRefill, 1);
 }
 
 
 bool
-ChunkCache::NeedsRefill()
+ChunkCache::SpaceLeft() const
 {
-	return atomic_or(&fNeedsRefill, 0);
+	ASSERT(IsLocked());
+
+	return fBytes < fMaxBytes;
 }
 
 
-status_t
+chunk_buffer*
-ChunkCache::GetNextChunk(const void** _chunkBuffer, size_t* _chunkSize,
+ChunkCache::NextChunk()
-	media_header* mediaHeader)
 {
-	uint8 retryCount = 0;
+	ASSERT(IsLocked());
 
-//	printf("ChunkCache::GetNextChunk: %p fEmptyChunkCount %ld, fReadyChunkCount %ld\n", fNextGet, fEmptyChunkCount, fReadyChunkCount);
+	chunk_buffer* chunk = fChunks.RemoveHead();
-retry:
+	if (chunk != NULL) {
-	acquire_sem(fGetWaitSem);
+		fBytes -= chunk->capacity;
-
+		fInFlightChunks.Add(chunk);
-	BAutolock locker(fLocker);
+		release_sem(fWaitSem);
-	if (fReadyChunkCount == 0) {
-		locker.Unlock();
-
-		printf("ChunkCache::GetNextChunk: %p retrying\n", fNextGet);
-		// Limit to 5 retries
-		retryCount++;
-		if (retryCount > 4)
-			return B_ERROR;
-
-		goto retry;
 	}
 
-	fEmptyChunkCount++;
+	return chunk;
-	fReadyChunkCount--;
-	atomic_or(&fNeedsRefill, 1);
-
-	locker.Unlock();
-
-	*_chunkBuffer = fNextGet->buffer;
-	*_chunkSize = fNextGet->sizeUsed;
-	*mediaHeader = fNextGet->mediaHeader;
-	status_t status = fNextGet->status;
-	fNextGet = fNextGet->next;
-
-	return status;
 }
 
 
+/*!	Moves the specified chunk from the in-flight list to the unused list.
+	This means the chunk data can be overwritten again.
+*/
 void
-ChunkCache::PutNextChunk(const void* chunkBuffer, size_t chunkSize,
+ChunkCache::RecycleChunk(chunk_buffer* chunk)
-	const media_header& mediaHeader, status_t status)
 {
-//	printf("ChunkCache::PutNextChunk: %p fEmptyChunkCount %ld, fReadyChunkCount %ld\n", fNextPut, fEmptyChunkCount, fReadyChunkCount);
+	ASSERT(IsLocked());
+
+	fInFlightChunks.Remove(chunk);
+	fUnusedChunks.Add(chunk);
+}
+
+
+bool
+ChunkCache::ReadNextChunk(Reader* reader, void* cookie)
+{
+	ASSERT(IsLocked());
+
+	// retrieve chunk buffer
+	chunk_buffer* chunk = fUnusedChunks.RemoveHead();
+	if (chunk == NULL) {
+		// allocate a new one
+		chunk = new(std::nothrow) chunk_buffer;
+		if (chunk == NULL)
+			return false;
 
-	if (status == B_OK) {
-		if (fNextPut->sizeMax < chunkSize) {
-//			printf("ChunkCache::PutNextChunk: %p resizing from %ld to %ld\n", fNextPut, fNextPut->sizeMax, chunkSize);
-			free(fNextPut->buffer);
-			fNextPut->buffer = malloc((chunkSize + 1024) & ~1023);
-			fNextPut->sizeMax = chunkSize;
-		}
-		memcpy(fNextPut->buffer, chunkBuffer, chunkSize);
-		fNextPut->sizeUsed = chunkSize;
 	}
 
-	fNextPut->mediaHeader = mediaHeader;
+	const void* buffer;
-	fNextPut->status = status;
+	size_t bufferSize;
+	chunk->status = reader->GetNextChunk(cookie, &buffer, &bufferSize,
+		&chunk->header);
+	if (chunk->status == B_OK) {
+		if (chunk->capacity < bufferSize) {
+			// adapt buffer size
+			free(chunk->buffer);
+			chunk->capacity = (bufferSize + 2047) & ~2047;
+			chunk->buffer = malloc(chunk->capacity);
+			if (chunk->buffer == NULL) {
+				delete chunk;
+				return false;
+			}
+		}
 
-	fNextPut = fNextPut->next;
+		memcpy(chunk->buffer, buffer, bufferSize);
+		chunk->size = bufferSize;
+		fBytes += chunk->capacity;
+	}
 
-	fLocker.Lock();
+	fChunks.Add(chunk);
-	fEmptyChunkCount--;
+	return chunk->status == B_OK;
-	fReadyChunkCount++;
-	if (fEmptyChunkCount == 0)
-		atomic_and(&fNeedsRefill, 0);
-	fLocker.Unlock();
-
-	release_sem(fGetWaitSem);
 }

src/kits/media/ChunkCache.h

@@ -1,5 +1,5 @@
 /*
- * Copyright 2004, Marcus Overhagen. All rights reserved.
+ * Copyright 2009, Axel Dörfler, [email protected].
  * Distributed under the terms of the MIT License.
  */
 #ifndef _CHUNK_CACHE_H
@@ -8,56 +8,54 @@
 
 #include <Locker.h>
 #include <MediaDefs.h>
+#include "ReaderPlugin.h"
+
+#include <kernel/util/DoublyLinkedList.h>
 
 
 namespace BPrivate {
 namespace media {
 
 
-struct chunk_info {
+struct chunk_buffer;
-	chunk_info*		next;
+typedef DoublyLinkedList<chunk_buffer> ChunkList;
+
+struct chunk_buffer : public DoublyLinkedListLinkImpl<chunk_buffer> {
+	chunk_buffer();
+	~chunk_buffer();
+
 	void*			buffer;
-	size_t			sizeUsed;
+	size_t			size;
-	size_t			sizeMax;
+	size_t			capacity;
-	media_header	mediaHeader;
+	media_header	header;
 	status_t		status;
 };
 
 
-class ChunkCache {
+class ChunkCache : public BLocker {
 public:
-								ChunkCache();
+								ChunkCache(sem_id waitSem, size_t maxBytes);
 								~ChunkCache();
 
 			void				MakeEmpty();
-			bool				NeedsRefill();
+			bool				SpaceLeft() const;
 
-			status_t			GetNextChunk(const void** _chunkBuffer,
+			chunk_buffer*		NextChunk();
-									size_t* _chunkSize,
+			void				RecycleChunk(chunk_buffer* chunk);
-									media_header* mediaHeader);
+			bool				ReadNextChunk(Reader* reader, void* cookie);
-			void				PutNextChunk(const void* chunkBuffer,
-									size_t chunkSize,
-									const media_header& mediaHeader,
-									status_t status);
 
 private:
-	enum { CHUNK_COUNT = 5 };
+			sem_id				fWaitSem;
-
+			size_t				fMaxBytes;
-			chunk_info*		fNextPut;
+			size_t				fBytes;
-			chunk_info*		fNextGet;
+			ChunkList			fChunks;
-			chunk_info		fChunkInfos[CHUNK_COUNT];
+			ChunkList			fUnusedChunks;
-
+			ChunkList			fInFlightChunks;
-			sem_id			fGetWaitSem;
-			int32			fEmptyChunkCount;
-			int32			fReadyChunkCount;
-			int32			fNeedsRefill;
-
-			BLocker			fLocker;
 };
 
 
-} // namespace media
+}	// namespace media
-} // namespace BPrivate
+}	// namespace BPrivate
 
 using namespace BPrivate::media;
 

src/kits/media/MediaExtractor.cpp

@@ -1,6 +1,8 @@
 /*
  * Copyright 2004-2007, Marcus Overhagen. All rights reserved.
  * Copyright 2008, Maurice Kalinowski. All rights reserved.
+ * Copyright 2009, Axel Dörfler, [email protected].
+ *
  * Distributed under the terms of the MIT License.
  */
 
@@ -13,8 +15,8 @@
 
 #include <Autolock.h>
 
-#include "debug.h"
 #include "ChunkCache.h"
+#include "debug.h"
 #include "PluginManager.h"
 
 
@@ -22,6 +24,9 @@
 #define DISABLE_CHUNK_CACHE 0
 
 
+static const size_t kMaxCacheBytes = 1024 * 1024;
+
+
 class MediaExtractorChunkProvider : public ChunkProvider {
 public:
 	MediaExtractorChunkProvider(MediaExtractor* extractor, int32 stream)
@@ -48,21 +53,28 @@ private:
 
 
 MediaExtractor::MediaExtractor(BDataIO* source, int32 flags)
+	:
+	fExtractorThread(-1),
+	fSource(source),
+	fReader(NULL),
+	fStreamInfo(NULL),
+	fStreamCount(0)
 {
 	CALLED();
-	fSource = source;
+
-	fStreamInfo = NULL;
+#if !DISABLE_CHUNK_CACHE
-	fExtractorThread = -1;
+	// start extractor thread
-	fExtractorWaitSem = -1;
+	fExtractorWaitSem = create_sem(1, "media extractor thread sem");
-	fTerminateExtractor = false;
+	if (fExtractorWaitSem < 0) {
+		fInitStatus = fExtractorWaitSem;
+		return;
+	}
+#endif
 
 	fInitStatus = _plugin_manager.CreateReader(&fReader, &fStreamCount,
 		&fFileFormat, source);
-	if (fInitStatus != B_OK) {
+	if (fInitStatus != B_OK)
-		fStreamCount = 0;
-		fReader = NULL;
 		return;
-	}
 
 	fStreamInfo = new stream_info[fStreamCount];
 
@@ -73,7 +85,8 @@ MediaExtractor::MediaExtractor(BDataIO* source, int32 flags)
 		fStreamInfo[i].hasCookie = true;
 		fStreamInfo[i].infoBuffer = 0;
 		fStreamInfo[i].infoBufferSize = 0;
-		fStreamInfo[i].chunkCache = new ChunkCache;
+		fStreamInfo[i].chunkCache
+			= new ChunkCache(fExtractorWaitSem, kMaxCacheBytes);
 		memset(&fStreamInfo[i].encodedFormat, 0,
 			sizeof(fStreamInfo[i].encodedFormat));
 	}
@@ -106,11 +119,10 @@ MediaExtractor::MediaExtractor(BDataIO* source, int32 flags)
 		}
 	}
 
-#if DISABLE_CHUNK_CACHE == 0
+#if !DISABLE_CHUNK_CACHE
 	// start extractor thread
-	fExtractorWaitSem = create_sem(1, "media extractor thread sem");
 	fExtractorThread = spawn_thread(_ExtractorEntry, "media extractor thread",
-		40, this);
+		B_NORMAL_PRIORITY + 4, this);
 	resume_thread(fExtractorThread);
 #endif
 }
@@ -120,24 +132,26 @@ MediaExtractor::~MediaExtractor()
 {
 	CALLED();
 
+#if !DISABLE_CHUNK_CACHE
 	// terminate extractor thread
-	fTerminateExtractor = true;
-	release_sem(fExtractorWaitSem);
-	status_t err;
-	wait_for_thread(fExtractorThread, &err);
 	delete_sem(fExtractorWaitSem);
 
+	status_t status;
+	wait_for_thread(fExtractorThread, &status);
+#endif
+
 	// free all stream cookies
 	// and chunk caches
 	for (int32 i = 0; i < fStreamCount; i++) {
 		if (fStreamInfo[i].hasCookie)
 			fReader->FreeCookie(fStreamInfo[i].cookie);
+
 		delete fStreamInfo[i].chunkCache;
 	}
 
 	_plugin_manager.DestroyReader(fReader);
 
-	delete [] fStreamInfo;
+	delete[] fStreamInfo;
 	// fSource is owned by the BMediaFile
 }
 
@@ -211,7 +225,7 @@ MediaExtractor::Duration(int32 stream) const
 	int64 frameCount;
 	bigtime_t duration;
 	media_format format;
-	const void *infoBuffer;
+	const void* infoBuffer;
 	size_t infoSize;
 
 	fReader->GetStreamInfo(fStreamInfo[stream].cookie, &frameCount, &duration,
@@ -222,20 +236,23 @@ MediaExtractor::Duration(int32 stream) const
 
 
 status_t
-MediaExtractor::Seek(int32 stream, uint32 seekTo, int64* frame, bigtime_t* time)
+MediaExtractor::Seek(int32 stream, uint32 seekTo, int64* _frame,
+	bigtime_t* _time)
 {
 	CALLED();
-	if (fStreamInfo[stream].status != B_OK)
-		return fStreamInfo[stream].status;
 
-	status_t result;
+	stream_info& info = fStreamInfo[stream];
-	result = fReader->Seek(fStreamInfo[stream].cookie, seekTo, frame, time);
+	if (info.status != B_OK)
-	if (result != B_OK)
+		return info.status;
-		return result;
 
-	// clear buffered chunks
+	BAutolock _(info.chunkCache);
-	fStreamInfo[stream].chunkCache->MakeEmpty();
+
-	release_sem(fExtractorWaitSem);
+	status_t status = fReader->Seek(info.cookie, seekTo, _frame, _time);
+	if (status != B_OK)
+		return status;
+
+	// clear buffered chunks after seek
+	info.chunkCache->MakeEmpty();
 
 	return B_OK;
 }
@@ -246,11 +263,21 @@ MediaExtractor::FindKeyFrame(int32 stream, uint32 seekTo, int64* _frame,
 	bigtime_t* _time) const
 {
 	CALLED();
-	if (fStreamInfo[stream].status != B_OK)
-		return fStreamInfo[stream].status;
 
-	return fReader->FindKeyFrame(fStreamInfo[stream].cookie,
+	stream_info& info = fStreamInfo[stream];
-		seekTo, _frame, _time);
+	if (info.status != B_OK)
+		return info.status;
+
+	BAutolock _(info.chunkCache);
+
+	status_t status = fReader->FindKeyFrame(info.cookie, seekTo, _frame, _time);
+	if (status != B_OK)
+		return status;
+
+	// clear buffered chunks after seek
+	info.chunkCache->MakeEmpty();
+
+	return B_OK;
 }
 
 
@@ -258,20 +285,41 @@ status_t
 MediaExtractor::GetNextChunk(int32 stream, const void** _chunkBuffer,
 	size_t* _chunkSize, media_header* mediaHeader)
 {
-	if (fStreamInfo[stream].status != B_OK)
+	stream_info& info = fStreamInfo[stream];
-		return fStreamInfo[stream].status;
 
-#if DISABLE_CHUNK_CACHE > 0
+	if (info.status != B_OK)
+		return info.status;
+
+#if DISABLE_CHUNK_CACHE
 	static BLocker locker("media extractor next chunk");
 	BAutolock lock(locker);
 	return fReader->GetNextChunk(fStreamInfo[stream].cookie, _chunkBuffer,
 		_chunkSize, mediaHeader);
-#endif
+#else
+	BAutolock _(info.chunkCache);
 
-	status_t status = fStreamInfo[stream].chunkCache->GetNextChunk(_chunkBuffer,
+	_RecycleLastChunk(info);
-		_chunkSize, mediaHeader);
+
-	release_sem(fExtractorWaitSem);
+	// Retrieve next chunk - read it directly, if the cache is drained
-	return status;
+	chunk_buffer* chunk;
+	do {
+		chunk = info.chunkCache->NextChunk();
+		if (chunk == NULL
+			&& !info.chunkCache->ReadNextChunk(fReader, info.cookie))
+			break;
+	} while (chunk == NULL);
+
+	if (chunk == NULL)
+		return B_NO_MEMORY;
+
+	info.lastChunk = chunk;
+
+	*_chunkBuffer = chunk->buffer;
+	*_chunkSize = chunk->size;
+	*mediaHeader = chunk->header;
+
+	return chunk->status;
+#endif
 }
 
 
@@ -339,6 +387,16 @@ MediaExtractor::CreateDecoder(int32 stream, Decoder** _decoder,
 }
 
 
+void
+MediaExtractor::_RecycleLastChunk(stream_info& info)
+{
+	if (info.lastChunk != NULL) {
+		info.chunkCache->RecycleChunk(info.lastChunk);
+		info.lastChunk = NULL;
+	}
+}
+
+
 status_t
 MediaExtractor::_ExtractorEntry(void* extractor)
 {
@@ -351,31 +409,35 @@ void
 MediaExtractor::_ExtractorThread()
 {
 	while (true) {
-		acquire_sem(fExtractorWaitSem);
+		status_t status;
-		if (fTerminateExtractor)
-			return;
-
-		bool refillDone;
 		do {
-			refillDone = false;
+			status = acquire_sem(fExtractorWaitSem);
-			for (int32 stream = 0; stream < fStreamCount; stream++) {
+		} while (status == B_INTERRUPTED);
-				if (fStreamInfo[stream].status != B_OK)
-					continue;
 
-				if (fStreamInfo[stream].chunkCache->NeedsRefill()) {
+		if (status != B_OK) {
-					media_header mediaHeader;
+			// we were asked to quit
-					const void* chunkBuffer;
+			return;
-					size_t chunkSize;
+		}
-					status_t status = fReader->GetNextChunk(
+
-						fStreamInfo[stream].cookie, &chunkBuffer, &chunkSize,
+		// Iterate over all streams until they are all filled
-						&mediaHeader);
+
-					fStreamInfo[stream].chunkCache->PutNextChunk(chunkBuffer,
+		int32 streamsFilled;
-						chunkSize, mediaHeader, status);
+		do {
-					refillDone = true;
+			streamsFilled = 0;
+
+			for (int32 stream = 0; stream < fStreamCount; stream++) {
+				stream_info& info = fStreamInfo[stream];
+				if (info.status != B_OK) {
+					streamsFilled++;
+					continue;
 				}
+
+				BAutolock _(info.chunkCache);
+
+				if (!info.chunkCache->SpaceLeft()
+					|| !info.chunkCache->ReadNextChunk(fReader, info.cookie))
+					streamsFilled++;
 			}
-			if (fTerminateExtractor)
+		} while (streamsFilled < fStreamCount);
-				return;
-		} while (refillDone);
 	}
 }