/*
 * Copyright 2008, Axel Dörfler, axeld@pinc-software.de.
 * Distributed under the terms of the MIT License.
 */


#include "AdaptiveBuffering.h"

#include <stdlib.h>


//#define TRACE(x...) printf(x)
#define TRACE(x...) ;


AdaptiveBuffering::AdaptiveBuffering(size_t initialBufferSize,
		size_t maxBufferSize, uint32 count)
	:
	fWriterThread(-1),
	fBuffers(NULL),
	fReadBytes(NULL),
	fBufferCount(count),
	fReadIndex(0),
	fWriteIndex(0),
	fReadCount(0),
	fWriteCount(0),
	fMaxBufferSize(maxBufferSize),
	fCurrentBufferSize(initialBufferSize),
	fReadSem(-1),
	fWriteSem(-1),
	fFinishedSem(-1),
	fWriteStatus(B_OK),
	fWriteTime(0),
	fFinished(false),
	fQuit(false)
{
}


AdaptiveBuffering::~AdaptiveBuffering()
{
	_QuitWriter();

	delete_sem(fReadSem);
	delete_sem(fWriteSem);

	if (fBuffers != NULL) {
		for (uint32 i = 0; i < fBufferCount; i++) {
			if (fBuffers[i] == NULL)
				break;

			free(fBuffers[i]);
		}

		free(fBuffers);
	}

	free(fReadBytes);
}


status_t
AdaptiveBuffering::Init()
{
	fReadBytes = (size_t*)malloc(fBufferCount * sizeof(size_t));
	if (fReadBytes == NULL)
		return B_NO_MEMORY;

	fBuffers = (uint8**)malloc(fBufferCount * sizeof(uint8*));
	if (fBuffers == NULL)
		return B_NO_MEMORY;

	for (uint32 i = 0; i < fBufferCount; i++) {
		fBuffers[i] = (uint8*)malloc(fMaxBufferSize);
		if (fBuffers[i] == NULL)
			return B_NO_MEMORY;
	}

	fReadSem = create_sem(0, "reader");
	if (fReadSem < B_OK)
		return fReadSem;

	fWriteSem = create_sem(fBufferCount - 1, "writer");
	if (fWriteSem < B_OK)
		return fWriteSem;

	fFinishedSem = create_sem(0, "finished");
	if (fFinishedSem < B_OK)
		return fFinishedSem;

	fWriterThread = spawn_thread(&_Writer, "buffer reader", B_LOW_PRIORITY,
		this);
	if (fWriterThread < B_OK)
		return fWriterThread;

	return resume_thread(fWriterThread);
}


status_t
AdaptiveBuffering::Read(uint8* /*buffer*/, size_t* _length)
{
	*_length = 0;
	return B_OK;
}


status_t
AdaptiveBuffering::Write(uint8* /*buffer*/, size_t /*length*/)
{
	return B_OK;
}


status_t
AdaptiveBuffering::Run()
{
	fReadIndex = 0;
	fWriteIndex = 0;
	fReadCount = 0;
	fWriteCount = 0;
	fWriteStatus = B_OK;
	fWriteTime = 0;

	while (fWriteStatus >= B_OK) {
		bigtime_t start = system_time();
		int32 index = fReadIndex;

		TRACE("%ld. read index %lu, buffer size %lu\n", fReadCount, index,
			fCurrentBufferSize);

		fReadBytes[index] = fCurrentBufferSize;
		status_t status = Read(fBuffers[index], &fReadBytes[index]);
		if (status < B_OK)
			return status;

		TRACE("%ld. read -> %lu bytes\n", fReadCount, fReadBytes[index]);

		fReadCount++;
		fReadIndex = (index + 1) % fBufferCount;
		if (fReadBytes[index] == 0)
			fFinished = true;
		release_sem(fReadSem);

		while (acquire_sem(fWriteSem) == B_INTERRUPTED)
			;

		if (fFinished)
			break;

		bigtime_t readTime = system_time() - start;
		uint32 writeTime = fWriteTime;
		if (writeTime) {
			if (writeTime > readTime) {
				fCurrentBufferSize = fCurrentBufferSize * 8/9;
				fCurrentBufferSize &= ~65535;
			} else {
				fCurrentBufferSize = fCurrentBufferSize * 9/8;
				fCurrentBufferSize = (fCurrentBufferSize + 65535) & ~65535;

				if (fCurrentBufferSize > fMaxBufferSize)
					fCurrentBufferSize = fMaxBufferSize;
			}
		}
	}

	while (acquire_sem(fFinishedSem) == B_INTERRUPTED)
		;

	return fWriteStatus;
}


void
AdaptiveBuffering::_QuitWriter()
{
	if (fWriterThread >= B_OK) {
		fQuit = true;
		release_sem(fReadSem);

		status_t status;
		wait_for_thread(fWriterThread, &status);

		fWriterThread = -1;
	}
}


status_t
AdaptiveBuffering::_Writer()
{
	while (true) {
		while (acquire_sem(fReadSem) == B_INTERRUPTED)
			;
		if (fQuit)
			break;

		bigtime_t start = system_time();

		TRACE("%ld. write index %lu, %p, bytes %lu\n", fWriteCount, fWriteIndex,
			fBuffers[fWriteIndex], fReadBytes[fWriteIndex]);

		fWriteStatus = Write(fBuffers[fWriteIndex], fReadBytes[fWriteIndex]);

		TRACE("%ld. write done\n", fWriteCount);

		fWriteIndex = (fWriteIndex + 1) % fBufferCount;
		fWriteTime = uint32(system_time() - start);
		fWriteCount++;

		release_sem(fWriteSem);

		if (fWriteStatus < B_OK)
			return fWriteStatus;
		if (fFinished && fWriteCount == fReadCount)
			release_sem(fFinishedSem);
	}

	return B_OK;
}


/*static*/ status_t
AdaptiveBuffering::_Writer(void* self)
{
	return ((AdaptiveBuffering*)self)->_Writer();
}