// Copyright (c) 2007-2009 Nokia Corporation and/or its subsidiary(-ies).
// All rights reserved.
// This component and the accompanying materials are made available
// under the terms of the License "Eclipse Public License v1.0"
// which accompanies this distribution, and is available
// at the URL "http://www.eclipse.org/legal/epl-v10.html".
//
// Initial Contributors:
// Nokia Corporation - initial contribution.
//
// Contributors:
//
// Description:
// e32test\nkernsa\fastbuf.cpp
// 
//

#include <nktest/nkutils.h>

template <class T>
class WaitFreePipe
	{
public:
	static WaitFreePipe<T>* New(TInt aSize);
	~WaitFreePipe();
	void InitReader();
	void InitWriter();
	void Read(T& aOut);
	TInt Write(const T& aIn);
	inline TUint32 Waits() {return iWaits;}
	inline void ResetWaits() {iWaits = 0;}
private:
	WaitFreePipe();
private:
	T* volatile iWrite;
	T* volatile iRead;
	T* iBase;
	T* iEnd;
	NRequestStatus* volatile iStat;
	NThread* iReader;
	volatile TUint32 iWaits;
	};

template <class T>
WaitFreePipe<T>::WaitFreePipe()
	:	iStat(0),
		iReader(0),
		iWaits(0)
	{
	}

template <class T>
WaitFreePipe<T>::~WaitFreePipe()
	{
	free(iBase);
	}

template <class T>
WaitFreePipe<T>* WaitFreePipe<T>::New(TInt aSize)
	{
	WaitFreePipe<T>* p = new WaitFreePipe<T>;
	if (!p)
		return 0;
	p->iBase = (T*)malloc(aSize * sizeof(T));
	if (!p->iBase)
		{
		delete p;
		return 0;
		}
	p->iEnd = p->iBase + aSize;
	p->iWrite = p->iBase;
	p->iRead = p->iBase;
	return p;
	}

template <class T>
void WaitFreePipe<T>::InitWriter()
	{
	}

template <class T>
void WaitFreePipe<T>::InitReader()
	{
	iReader = NKern::CurrentThread();
	}

template <class T>
void WaitFreePipe<T>::Read(T& aOut)
	{
	while (iRead == iWrite)
		{
		NRequestStatus s;
		s = KRequestPending;
		// make sure set to KRequestPending is seen before iStat write
		__e32_atomic_store_ord_ptr(&iStat, &s);
		// make sure writer sees our request status before we check for buffer empty again
		if (iRead != iWrite)
			RequestComplete(iReader, (NRequestStatus*&)iStat, 0);
		WaitForRequest(s);
		++iWaits;
		}
	aOut = *iRead;
	T* new_read = iRead + 1;
	if (new_read == iEnd)
		new_read = iBase;
	// make sure read of data value is observed before update of read pointer
	__e32_atomic_store_rel_ptr(&iRead, new_read);
	}

template <class T>
TInt WaitFreePipe<T>::Write(const T& aIn)
	{
	T* new_write = iWrite + 1;
	if (new_write == iEnd)
		new_write = iBase;
	if (new_write == iRead)
		return KErrOverflow;	// buffer full
	*iWrite = aIn;
	// make sure data is seen before updated write pointer
	__e32_atomic_store_ord_ptr(&iWrite, new_write);
	if (iStat)
		RequestComplete(iReader, (NRequestStatus*&)iStat, 0);
	return KErrNone;
	}


struct SPipeTest
	{
	WaitFreePipe<TUint32>* iPipe;
	TUint64 iTotalWrites;
	TUint64 iTotalReads;
	volatile TUint32 iWrites;
	volatile TUint32 iReads;
	TUint32 iMeasure;
	volatile TUint32 iReadTime;
	volatile TUint32 iWriteTime;
	volatile TBool iStop;
	};

void PipeWriterThread(TAny* aPtr)
	{
	SPipeTest& a = *(SPipeTest*)aPtr;
	a.iPipe->InitWriter();
	TUint32 seed[2] = {1,0};
	TUint32 seqs[2] = {3,0};
	TInt r;
	while (!a.iStop)
		{
		TUint32 x = random(seqs);
		do	{
			r = a.iPipe->Write(x);
			if (r != KErrNone)
				fcfspin(2*a.iWriteTime);
			} while (r != KErrNone);
		++a.iTotalWrites;
		++a.iWrites;
		while (a.iWrites>=a.iMeasure)
			{}
		TUint32 time = random(seed) % a.iWriteTime;
		fcfspin(time);
		}
	}

void PipeReaderThread(TAny* aPtr)
	{
	SPipeTest& a = *(SPipeTest*)aPtr;
	TUint32 seed[2] = {2,0};
	TUint32 seqs[2] = {3,0};
	a.iPipe->InitReader();
	a.iPipe->ResetWaits();
	while (!a.iStop)
		{
		TUint32 x = random(seqs);
		TUint32 y;
		a.iPipe->Read(y);
		TEST_RESULT(x==y, "Wrong value");
		++a.iTotalReads;
		++a.iReads;
		if (a.iReads < a.iMeasure)
			{
			TUint32 time = random(seed) % a.iReadTime;
			fcfspin(time);
			continue;
			}
		TUint32 w = a.iPipe->Waits();
		TUint32 wr = (w<<4)/a.iMeasure;
		TEST_PRINT3("%d waits out of %d (wr=%d)", w, a.iMeasure, wr);
		TUint32 rt = a.iReadTime;
		TUint32 wt = a.iWriteTime;
		switch (wr)
			{
			case 0:
				a.iReadTime = rt>>1;
				a.iWriteTime = wt<<1;
				break;
			case 1:
			case 2:
			case 3:
				a.iReadTime = rt - (rt>>2);
				a.iWriteTime = wt + (wt>>2);
				break;
			case 4:
			case 5:
			case 6:
				a.iReadTime = rt - (rt>>3);
				a.iWriteTime = wt + (wt>>3);
				break;
			case 7:
			case 8:
				// ok
				break;
			case 9:
			case 10:
			case 11:
				a.iReadTime = rt - (rt>>3);
				a.iWriteTime = wt + (wt>>3);
				break;
			case 12:
			case 13:
			case 14:
				a.iReadTime = rt + (rt>>2);
				a.iWriteTime = wt - (wt>>2);
				break;
			case 15:
			case 16:
				a.iReadTime = rt<<1;
				a.iWriteTime = wt>>1;
				break;
			}
		TEST_PRINT4("RT: %d->%d WT: %d->%d", rt, a.iReadTime, wt, a.iWriteTime);
		a.iPipe->ResetWaits();
		a.iReads = 0;
		a.iWrites = 0;
		}
	}

void DoPipeTest()
	{
	SPipeTest a;
	memclr(&a, sizeof(a));
	a.iPipe = WaitFreePipe<TUint32>::New(1024);
	TEST_OOM(a.iPipe);
	a.iMeasure = 131072;
	a.iReadTime = 1024;
	a.iWriteTime = 1024;

	NFastSemaphore exitSem(0);
	NThread* reader = CreateThreadSignalOnExit("Reader", &PipeReaderThread, 11, &a, 0, -1, &exitSem, 0);
	TEST_OOM(reader);
	NThread* writer = CreateThreadSignalOnExit("Writer", &PipeWriterThread, 11, &a, 0, -1, &exitSem, 1);
	TEST_OOM(writer);

	while (a.iTotalWrites < 0x01000000u)
		NKern::Sleep(1000);
	a.iStop = TRUE;

	NKern::FSWait(&exitSem);
	NKern::FSWait(&exitSem);
	}

void TestWaitFreePipe()
	{
	DoPipeTest();
	}