// Copyright (c) 2008-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\misc\t_svrstress.cpp

// This is a stress test for client server session connect and disconnect

//

#include <e32base.h>

#include <e32base_private.h>

#define __E32TEST_EXTENSION__

#include <e32test.h>

#include <e32svr.h>

#include "u32std.h"

#include <e32atomics.h>

#include <e32panic.h>

#include <e32def.h>

#include <e32def_private.h>

RTest test(_L("T_SVRSTRESS"));

RSemaphore SyncSemaphore;

TUint32 WaitABit;

TInt NumMessageSlots;

TInt UseGlobalMessagePool;

const TInt BigDesLength = 256 * 1024;

#if 1

#define TRACE(t) RDebug::RawPrint(_L8(t))

#else

#define TRACE(t)

#endif

//

// utility functions...

//

void WaitForRequest()

	{

	User::WaitForAnyRequest();

	RThread().RequestSignal(); // put request semaphore count back

	}

TInt WaitForRequest(TRequestStatus& aStatus,TTimeIntervalMicroSeconds32 aTimeout=2*1000000)

	{

	RTimer timer;

	test_Equal(KErrNone,timer.CreateLocal());

	TRequestStatus timeoutStatus;

	timer.After(timeoutStatus,aTimeout);

	User::WaitForRequest(aStatus,timeoutStatus);

	TInt r;

	if(aStatus.Int()==KRequestPending)

		{

		r = KErrTimedOut;

		}

	else

		{

		r = KErrNone;

		timer.Cancel();

		User::WaitForRequest(timeoutStatus);

		}

	CLOSE_AND_WAIT(timer);

	return r;

	}

//

// CMyServer

//

_LIT(KMyServerName,"StressSvr");

class CMyServer : public CServer2

	{

public:

	CMyServer(TInt aPriority);

	static CMyServer* New(TInt aPriority);

	virtual CSession2* NewSessionL(const TVersion&, const RMessage2&) const;

	};

class CMySession : public CSession2

	{

public:

	virtual void ServiceL(const RMessage2& aMessage);

	};

CMyServer* CMyServer::New(TInt aPriority)

	{

	return new CMyServer(aPriority);

	}

CMyServer::CMyServer(TInt aPriority)

	: CServer2(aPriority, ESharableSessions)

	{}

CSession2* CMyServer::NewSessionL(const TVersion&, const RMessage2&) const

	{

	TRACE("O");

	return new(ELeave) CMySession;

	}

TBool RestartServer;

TInt MyServerThread(TAny*)

	{

	CActiveScheduler* pR=new CActiveScheduler;

	if(!pR)

		return KErrNoMemory;

	CActiveScheduler::Install(pR);

	RestartServer = ETrue;

	while(RestartServer)

		{

		__UHEAP_MARK;

		CMyServer* pS=CMyServer::New(0);

		if(!pS)

			return KErrNoMemory;

		TInt r = pS->Start(KMyServerName);

		if(r!=KErrNone)

			return r;

		TRACE("S");

		RThread::Rendezvous(KErrNone);

		CActiveScheduler::Start();

		delete pS;

		__UHEAP_MARKEND;

		}

	delete pR;

	return KErrNone;

	}

//

// RMyServer

//

class RMyServer : public RSessionBase

	{

public:

	enum TFunction

		{

		EStop,

		ESync,

		EPing,

		EShutdown,

		ECompleteWhileCopying

		};

public:

	TInt Connect();

	inline TInt Send(TFunction aFunction) const

		{ return SendReceive(aFunction); }

	inline TInt Send(TFunction aFunction, const TIpcArgs& aArgs) const

		{ return SendReceive(aFunction, aArgs); }

	inline void Send(TFunction aFunction, TRequestStatus& aStatus) const

		{ SendReceive(aFunction, aStatus); }

	inline void Send(TFunction aFunction, const TIpcArgs& aArgs, TRequestStatus& aStatus) const

		{ SendReceive(aFunction, aArgs, aStatus); }

	};

TInt RMyServer::Connect()

	{

	RMyServer temp;

	TInt r = temp.CreateSession(KMyServerName, TVersion(), UseGlobalMessagePool ? -1 : NumMessageSlots);

	if(r!=KErrNone)

		return r;

	// turn handle into process owned...

	RMyServer temp2(temp);

	r = temp2.Duplicate(RThread());

	temp.Close();

	*this = temp2;

	return r;

	}

//

// CMySession

//

TInt CopierThread(TAny* aPtr)

	{

	RMessage2& msg = *(RMessage2*)aPtr;

	HBufC* bigdes = HBufC::NewMax(BigDesLength);

	if (bigdes == NULL)

		return KErrNoMemory;

	TPtr ptr = bigdes->Des();

	RThread().Rendezvous(KErrNone);

	RDebug::Print(_L("START\n"));

	TInt r = msg.Read(2, ptr);

	RDebug::Print(_L("DONE\n"));

	delete bigdes;

	return r;

	}

void CMySession::ServiceL(const RMessage2& aMessage)

	{

	RThread client;

	RThread copier;

	aMessage.Client(client);

	TRequestStatus* s;

	TRequestStatus* s2;

	TRequestStatus logon, rendez;

	TInt r;

	s = (TRequestStatus*)aMessage.Ptr0();

	switch(aMessage.Function())

		{

	case RMyServer::EStop:

		TRACE("E");

		CActiveScheduler::Stop();

		break;

	case RMyServer::ESync:

		TRACE("Y");

		client.RequestComplete(s,KErrNone);		// let client know we've received the message

		SyncSemaphore.Wait();					// wait for signal from client

		s = (TRequestStatus*)aMessage.Ptr1();	// use second status for later end signal

		aMessage.Complete(KErrNone);			// complete the message

		break;

	case RMyServer::EPing:

		TRACE("P");

		aMessage.Complete(KErrNone);

		break;

	case RMyServer::EShutdown:

		TRACE("D");

		RestartServer = EFalse;

		CActiveScheduler::Stop();

		break;

	case RMyServer::ECompleteWhileCopying:

		s2 = (TRequestStatus*)aMessage.Ptr1();

		r = copier.Create(_L("Copier"),CopierThread,KDefaultStackSize,&User::Allocator(),(TAny*)&aMessage);

		if (r == KErrNone)

			{

			copier.Logon(logon);

			copier.Rendezvous(rendez);

			copier.SetPriority(EPriorityLess);

			copier.Resume();

			User::WaitForRequest(rendez);

			User::AfterHighRes(5000); // 5ms delay to let copy actually start

			RDebug::Print(_L("COMPLETING\n"));

			aMessage.Complete(KErrNone);

			User::WaitForRequest(logon);

			copier.Close();

			}

		client.RequestComplete(s,r);

		s = s2;

		break;

	default:

		TRACE("?");

		aMessage.Complete(KErrNotSupported);

		break;

		}

	// let client know we've completed the message...

	TRACE("X");

	client.RequestComplete(s,KErrNone);

	client.Close();

	}

//

// RStressThread

//

class RStressThread

	{

public:

	RStressThread(TThreadFunction aThreadFunction, const char* aName, TInt aDelay=-1);

	~RStressThread();

	void Start();

	void Restart();

	void Stop();

	// for use by thread...

	static RStressThread& Begin(TAny* aInfo);

	TBool Loop();

private:

	TThreadFunction iThreadFunction;

	const char* iName;

	RThread iThread;

	TRequestStatus iLogon;

	TUint iCount;

	TBool iStop;

	TInt iDelay;

private:

	static TInt iInstanceCounter;

	};

TInt RStressThread::iInstanceCounter = 0;

RStressThread::RStressThread(TThreadFunction aThreadFunction, const char* aName, TInt aDelay)

	: iThreadFunction(aThreadFunction), iName(aName), iLogon(KErrNone), iDelay(aDelay)

	{

	iThread.SetHandle(0);

	}

RStressThread::~RStressThread()

	{

	Stop();

	}

void RStressThread::Start()

	{

	iStop = false;

	iCount = 0;

	TBuf<KMaxKernelName> name;

	name.Copy(TPtrC8((const TUint8*)iName));

	name.Append((TText)'-');

	name.AppendNum(iInstanceCounter++);

	test_Equal(KErrNone,iThread.Create(name,iThreadFunction,KDefaultStackSize,&User::Allocator(),this));

	iThread.Logon(iLogon);

	test_Equal(KRequestPending,iLogon.Int());

	TRequestStatus rendezvous;

	iThread.Rendezvous(rendezvous);

	iThread.Resume();

	User::WaitForRequest(rendezvous);

	test_Equal(KErrNone,rendezvous.Int());

	}

void RStressThread::Stop()

	{

	if(!iThread.Handle())

		return; // thread not running

	iStop = true;

	RDebug::Printf("RStressThread::Stop %s (count=%d)",iName,iCount);

	if(WaitForRequest(iLogon,10*1000000)!=KErrNone)

		test(0);

	CLOSE_AND_WAIT(iThread);

	}

void RStressThread::Restart()

	{

	if(iThread.Handle())

		{

		if(iLogon==KRequestPending)

			return; // thread still running

		User::WaitForRequest(iLogon);

		CLOSE_AND_WAIT(iThread);

		}

	Start();

	}

TBool RStressThread::Loop()

	{

	if(iDelay>=0)

		User::AfterHighRes(iDelay);

	++iCount;

	return !iStop;

	}

RStressThread& RStressThread::Begin(TAny* aInfo)

	{

	RStressThread& t = *(RStressThread*)aInfo;

	if(t.iDelay>=0)

		RThread().SetPriority(EPriorityMore); // so this preempts threads after delay

	RThread::Rendezvous(KErrNone);

	return t;

	}

//

//

//

RMyServer Session;

RThread ServerThread;

void NewSession()

	{

	RMyServer newSession;

	TRACE("o");

	test_Equal(KErrNone,newSession.Connect());

	RMyServer oldSession(Session);

	Session = newSession;

	TRACE("c");

	if(oldSession.Handle())

		CLOSE_AND_WAIT(oldSession);

	}

TInt SessionCloserThread(TAny* aInfo)

	{

	RStressThread& t = RStressThread::Begin(aInfo);

	do

		{

		NewSession();

		}

	while(t.Loop());

	return KErrNone;

	}

TInt ServerStopperThread(TAny* aInfo)

	{

	RStressThread& t = RStressThread::Begin(aInfo);

	do

		{

		TRACE("s");

		TRequestStatus rendezvous;

		ServerThread.Rendezvous(rendezvous);

		TRequestStatus s1 = KRequestPending;

		TRequestStatus s2;

		Session.Send(RMyServer::EStop,TIpcArgs(&s1),s2);

		User::WaitForRequest(s1,s2);

		if(s2!=KRequestPending)

			{

			test_Equal(KErrServerTerminated,s2.Int());

			User::WaitForRequest(s1);

			}

		User::WaitForRequest(rendezvous);

		NewSession();

		}

	while(t.Loop());

	return KErrNone;

	}

TInt SessionPingerThread(TAny* aInfo)

	{

	RStressThread& t = RStressThread::Begin(aInfo);

	do

		{

		TRACE("p");

		TRequestStatus s1 = KRequestPending;

		TRequestStatus s2;

		Session.Send(RMyServer::EPing,TIpcArgs(&s1),s2);

		User::WaitForRequest(s1,s2);

		if(s2.Int()==KErrNone)

			{

			// message completed OK, wait for servers extra signal

			User::WaitForRequest(s1);

			}

		else if(s2.Int()==KErrServerTerminated)

			{

			// server died before message processed, there shouldn't be an extra signal

			test_Equal(KRequestPending,s1.Int());

			}

		else

			{

			// assume message was completed by server, but we didn't get signalled because session was closed

			test_Equal(KRequestPending,s2.Int());

			test_Equal(KErrNone,s1.Int());

			}

		}

	while(t.Loop());

	return KErrNone;

	}

void TestInit()

	{

	RThread().SetPriority(EPriorityMuchMore); // so this main thread is higher priority than workers

	test_Equal(KErrNone,SyncSemaphore.CreateLocal(0,EOwnerProcess));

	// calculate async cleanup timeout value...

	TInt factor = UserSvr::HalFunction(EHalGroupVariant, EVariantHalTimeoutExpansion, 0, 0);

	if (factor<=0)

		factor = 1;

	if (factor>1024)

		factor = 1024;

	WaitABit = 200000 * (TUint32)factor;

	}

void StartServer()

	{

	// start test server...

	test_Equal(KErrNone,ServerThread.Create(_L("Server"),MyServerThread,KDefaultStackSize,1<<12,1<<20,0));

	TRequestStatus rendezvous;

	ServerThread.Rendezvous(rendezvous);

	ServerThread.Resume();

	User::WaitForRequest(rendezvous);

	test_Equal(KErrNone,rendezvous.Int());

	test_Equal(EExitPending,ServerThread.ExitType());

	}

void StopServer()

	{

	TRequestStatus logon;

	NewSession();

	TRequestStatus s1 = KRequestPending;

	TRequestStatus s2;

	ServerThread.Logon(logon);

	Session.Send(RMyServer::EShutdown,TIpcArgs(&s1),s2);

	User::WaitForRequest(s1,s2);

	if(s2!=KRequestPending)

		{

		test_Equal(KErrServerTerminated,s2.Int());

		User::WaitForRequest(s1);

		}

	CLOSE_AND_WAIT(Session);

	User::WaitForRequest(logon);

	test_KErrNone(logon.Int());

	test_Equal(EExitKill, ServerThread.ExitType());

	CLOSE_AND_WAIT(ServerThread);

	}

void TestMessageCompleteOnClosedSession()

	{

	__KHEAP_MARK;

	test.Start(_L("Start server"));

	StartServer();

	test.Next(_L("Connect"));

	test_Equal(KErrNone,Session.Connect());

	test.Next(_L("Send message"));

	TRequestStatus s1 = KRequestPending;

	TRequestStatus s2 = KRequestPending;

	TRequestStatus s3;

	Session.Send(RMyServer::ESync,TIpcArgs(&s1,&s2),s3);

	test_Equal(KRequestPending,s3.Int());

	test.Next(_L("Wait for s1"));

	test_Equal(KErrNone,WaitForRequest(s1));

	test_Equal(KErrNone,s1.Int());

	test_Equal(KRequestPending,s2.Int());

	test_Equal(KRequestPending,s3.Int());

	test.Next(_L("Close session"));

	Session.Close();

	test_Equal(KRequestPending,s2.Int());

	test_Equal(KRequestPending,s3.Int());

	test.Next(_L("Trigger message completion"));

	SyncSemaphore.Signal();

	test.Next(_L("Wait for s2"));

	test_Equal(KErrNone,WaitForRequest(s2));

	test_Equal(KErrNone,s2.Int());

	test_Equal(KRequestPending,s3.Int());

	test.Next(_L("Stop server"));

	StopServer();

	test.End();

	User::After(WaitABit);	// allow asynchronous cleanup to happen

	__KHEAP_MARKEND;

	}

void TestMessageCompleteWhileCopying()

	{

	__KHEAP_MARK;

	test.Start(_L("Start server"));

	StartServer();

	test.Next(_L("Connect"));

	test_Equal(KErrNone,Session.Connect());

	test.Next(_L("Create large descriptor"));

	HBufC* bigdes = HBufC::NewMax(BigDesLength);

	test_NotNull(bigdes);

	TPtr ptr = bigdes->Des();

	test.Next(_L("Send message"));

	TRequestStatus s1 = KRequestPending;

	TRequestStatus s2 = KRequestPending;

	TRequestStatus s3;

	Session.Send(RMyServer::ECompleteWhileCopying,TIpcArgs(&s1,&s2,&ptr),s3);

	test.Next(_L("Wait for s3"));

	test_Equal(KErrNone,WaitForRequest(s3,10*1000000));

	test_Equal(KErrNone,s3.Int());

	test.Next(_L("Wait for s2"));

	test_Equal(KErrNone,WaitForRequest(s2,10*1000000));

	test_Equal(KErrNone,s2.Int());

	test.Next(_L("Wait for s1"));

	test_Equal(KErrNone,WaitForRequest(s1,10*1000000));

	test_Equal(KErrNone,s1.Int());

	test.Next(_L("Close session"));

	Session.Close();

	test.Next(_L("Stop server"));

	StopServer();

	test.End();

	User::After(WaitABit);	// allow asynchronous cleanup to happen

	__KHEAP_MARKEND;

	}

void RunStressThreads(RStressThread& aThread1, RStressThread& aThread2, TInt aTimeout=1000000)

	{

	__KHEAP_MARK;

	StartServer();

	NewSession();

	aThread1.Start();

	aThread2.Start();

	RTimer timer;

	test_Equal(KErrNone,timer.CreateLocal());

	TRequestStatus timeoutStatus;

	timer.After(timeoutStatus,aTimeout);

	do

		{

		aThread1.Restart();

		aThread2.Restart();

		WaitForRequest();

		}

	while(timeoutStatus==KRequestPending);

	User::WaitForRequest(timeoutStatus);

	CLOSE_AND_WAIT(timer);

	aThread2.Stop();

	aThread1.Stop();

	CLOSE_AND_WAIT(Session);

	StopServer();

	User::After(WaitABit);	// allow asynchronous cleanup to happen

	__KHEAP_MARKEND;

	}

GLDEF_C TInt E32Main()

	{

	TInt i;

	test.Title();

	test.Start(_L("Initialise"));

	TestInit();

	for(UseGlobalMessagePool=0; UseGlobalMessagePool<2; ++UseGlobalMessagePool)

		{

		if(UseGlobalMessagePool)

			test.Next(_L("Tests using global message pool"));

		else

			test.Next(_L("Tests using local message pool"));

		NumMessageSlots = 1;

		test.Start(_L("Check completing messages on dead session"));

		TestMessageCompleteOnClosedSession();

		for (i=0; i<10; i++)

			{

			test.Next(_L("Check completing message while IPC copying"));

			TestMessageCompleteWhileCopying();

			}

		test.Next(_L("Stress closing session whilst in use"));

		RStressThread closer(SessionCloserThread,"SessionCloser",0);

		RStressThread pinger1(SessionPingerThread,"Pinger");

		RunStressThreads(closer, pinger1);

		NumMessageSlots = 2;

		test.Next(_L("Stress stopping server whilst in use"));

		RStressThread stopper(ServerStopperThread,"ServerStopper",0);

		RStressThread pinger2(SessionPingerThread,"Pinger");

		RunStressThreads(stopper, pinger2);

		test.End();

		}

	test.End();

	return(0);

	}