executor/xeTcpIpLink.cpp

/*-------------------------------------------------------------------------
 * drawElements Quality Program Test Executor
 * ------------------------------------------
 *
 * Copyright 2014 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *//*!
 * \file
 * \brief Tcp/Ip communication link.
 *//*--------------------------------------------------------------------*/

#include "xeTcpIpLink.hpp"
#include "xsProtocol.hpp"
#include "deClock.h"
#include "deInt32.h"

namespace xe
{

enum
{
	SEND_BUFFER_BLOCK_SIZE		= 1024,
	SEND_BUFFER_NUM_BLOCKS		= 64
};

// Utilities for writing messages out.

static void writeMessageHeader (de::BlockBuffer<deUint8>& dst, xs::MessageType type, int messageSize)
{
	deUint8 hdr[xs::MESSAGE_HEADER_SIZE];
	xs::Message::writeHeader(type, messageSize, &hdr[0], xs::MESSAGE_HEADER_SIZE);
	dst.write(xs::MESSAGE_HEADER_SIZE, &hdr[0]);
}

static void writeKeepalive (de::BlockBuffer<deUint8>& dst)
{
	writeMessageHeader(dst, xs::MESSAGETYPE_KEEPALIVE, xs::MESSAGE_HEADER_SIZE);
	dst.flush();
}

static void writeExecuteBinary (de::BlockBuffer<deUint8>& dst, const char* name, const char* params, const char* workDir, const char* caseList)
{
	int		nameSize			= (int)strlen(name)		+ 1;
	int		paramsSize			= (int)strlen(params)	+ 1;
	int		workDirSize			= (int)strlen(workDir)	+ 1;
	int		caseListSize		= (int)strlen(caseList)	+ 1;
	int		totalSize			= xs::MESSAGE_HEADER_SIZE + nameSize + paramsSize + workDirSize + caseListSize;

	writeMessageHeader(dst, xs::MESSAGETYPE_EXECUTE_BINARY, totalSize);
	dst.write(nameSize,		(const deUint8*)name);
	dst.write(paramsSize,	(const deUint8*)params);
	dst.write(workDirSize,	(const deUint8*)workDir);
	dst.write(caseListSize,	(const deUint8*)caseList);
	dst.flush();
}

static void writeStopExecution (de::BlockBuffer<deUint8>& dst)
{
	writeMessageHeader(dst, xs::MESSAGETYPE_STOP_EXECUTION, xs::MESSAGE_HEADER_SIZE);
	dst.flush();
}

// TcpIpLinkState

TcpIpLinkState::TcpIpLinkState (CommLinkState initialState, const char* initialErr)
	: m_state					(initialState)
	, m_error					(initialErr)
	, m_lastKeepaliveReceived	(0)
	, m_stateChangedCallback	(DE_NULL)
	, m_testLogDataCallback		(DE_NULL)
	, m_infoLogDataCallback		(DE_NULL)
	, m_userPtr					(DE_NULL)
{
}

TcpIpLinkState::~TcpIpLinkState (void)
{
}

CommLinkState TcpIpLinkState::getState (void) const
{
	de::ScopedLock lock(m_lock);

	return m_state;
}

CommLinkState TcpIpLinkState::getState (std::string& error) const
{
	de::ScopedLock lock(m_lock);

	error = m_error;
	return m_state;
}

void TcpIpLinkState::setCallbacks (CommLink::StateChangedFunc stateChangedCallback, CommLink::LogDataFunc testLogDataCallback, CommLink::LogDataFunc infoLogDataCallback, void* userPtr)
{
	de::ScopedLock lock(m_lock);

	m_stateChangedCallback		= stateChangedCallback;
	m_testLogDataCallback		= testLogDataCallback;
	m_infoLogDataCallback		= infoLogDataCallback;
	m_userPtr					= userPtr;
}

void TcpIpLinkState::setState (CommLinkState state, const char* error)
{
	CommLink::StateChangedFunc	callback	= DE_NULL;
	void*						userPtr		= DE_NULL;

	{
		de::ScopedLock lock(m_lock);

		m_state = state;
		m_error	= error;

		callback	= m_stateChangedCallback;
		userPtr		= m_userPtr;
	}

	if (callback)
		callback(userPtr, state, error);
}

void TcpIpLinkState::onTestLogData (const deUint8* bytes, size_t numBytes) const
{
	CommLink::LogDataFunc	callback	= DE_NULL;
	void*					userPtr		= DE_NULL;

	m_lock.lock();
	callback	= m_testLogDataCallback;
	userPtr		= m_userPtr;
	m_lock.unlock();

	if (callback)
		callback(userPtr, bytes, numBytes);
}

void TcpIpLinkState::onInfoLogData (const deUint8* bytes, size_t numBytes) const
{
	CommLink::LogDataFunc	callback	= DE_NULL;
	void*					userPtr		= DE_NULL;

	m_lock.lock();
	callback	= m_infoLogDataCallback;
	userPtr		= m_userPtr;
	m_lock.unlock();

	if (callback)
		callback(userPtr, bytes, numBytes);
}

void TcpIpLinkState::onKeepaliveReceived (void)
{
	de::ScopedLock lock(m_lock);
	m_lastKeepaliveReceived = deGetMicroseconds();
}

deUint64 TcpIpLinkState::getLastKeepaliveRecevied (void) const
{
	de::ScopedLock lock(m_lock);
	return m_lastKeepaliveReceived;
}

// TcpIpSendThread

TcpIpSendThread::TcpIpSendThread (de::Socket& socket, TcpIpLinkState& state)
	: m_socket		(socket)
	, m_state		(state)
	, m_buffer		(SEND_BUFFER_BLOCK_SIZE, SEND_BUFFER_NUM_BLOCKS)
	, m_isRunning	(false)
{
}

TcpIpSendThread::~TcpIpSendThread (void)
{
}

void TcpIpSendThread::start (void)
{
	DE_ASSERT(!m_isRunning);

	// Reset state.
	m_buffer.clear();
	m_isRunning = true;

	de::Thread::start();
}

void TcpIpSendThread::run (void)
{
	try
	{
		deUint8 buf[SEND_BUFFER_BLOCK_SIZE];

		while (!m_buffer.isCanceled())
		{
			size_t			numToSend	= 0;
			size_t			numSent		= 0;
			deSocketResult	result		= DE_SOCKETRESULT_LAST;

			try
			{
				// Wait for single byte and then try to read more.
				m_buffer.read(1, &buf[0]);
				numToSend = 1 + m_buffer.tryRead(DE_LENGTH_OF_ARRAY(buf)-1, &buf[1]);
			}
			catch (const de::BlockBuffer<deUint8>::CanceledException&)
			{
				// Handled in loop condition.
			}

			while (numSent < numToSend)
			{
				result = m_socket.send(&buf[numSent], numToSend-numSent, &numSent);

				if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
					XE_FAIL("Connection closed");
				else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
					XE_FAIL("Connection terminated");
				else if (result == DE_SOCKETRESULT_ERROR)
					XE_FAIL("Socket error");
				else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
				{
					// \note Socket should not be in non-blocking mode.
					DE_ASSERT(numSent == 0);
					deYield();
				}
				else
					DE_ASSERT(result == DE_SOCKETRESULT_SUCCESS);
			}
		}
	}
	catch (const std::exception& e)
	{
		m_state.setState(COMMLINKSTATE_ERROR, e.what());
	}
}

void TcpIpSendThread::stop (void)
{
	if (m_isRunning)
	{
		m_buffer.cancel();
		join();
		m_isRunning = false;
	}
}

// TcpIpRecvThread

TcpIpRecvThread::TcpIpRecvThread (de::Socket& socket, TcpIpLinkState& state)
	: m_socket		(socket)
	, m_state		(state)
	, m_curMsgPos	(0)
	, m_isRunning	(false)
{
}

TcpIpRecvThread::~TcpIpRecvThread (void)
{
}

void TcpIpRecvThread::start (void)
{
	DE_ASSERT(!m_isRunning);

	// Reset state.
	m_curMsgPos = 0;
	m_isRunning = true;

	de::Thread::start();
}

void TcpIpRecvThread::run (void)
{
	try
	{
		for (;;)
		{
			bool				hasHeader		= m_curMsgPos >= xs::MESSAGE_HEADER_SIZE;
			bool				hasPayload		= false;
			size_t				messageSize		= 0;
			xs::MessageType		messageType		= (xs::MessageType)0;

			if (hasHeader)
			{
				xs::Message::parseHeader(&m_curMsgBuf[0], xs::MESSAGE_HEADER_SIZE, messageType, messageSize);
				hasPayload = m_curMsgPos >= messageSize;
			}

			if (hasPayload)
			{
				// Process message.
				handleMessage(messageType, m_curMsgPos > xs::MESSAGE_HEADER_SIZE ? &m_curMsgBuf[xs::MESSAGE_HEADER_SIZE] : DE_NULL, messageSize-xs::MESSAGE_HEADER_SIZE);
				m_curMsgPos = 0;
			}
			else
			{
				// Try to receive missing bytes.
				size_t				curSize			= hasHeader ? messageSize : (size_t)xs::MESSAGE_HEADER_SIZE;
				size_t				bytesToRecv		= curSize-m_curMsgPos;
				size_t				numRecv			= 0;
				deSocketResult		result			= DE_SOCKETRESULT_LAST;

				if (m_curMsgBuf.size() < curSize)
					m_curMsgBuf.resize(curSize);

				result = m_socket.receive(&m_curMsgBuf[m_curMsgPos], bytesToRecv, &numRecv);

				if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
					XE_FAIL("Connection closed");
				else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
					XE_FAIL("Connection terminated");
				else if (result == DE_SOCKETRESULT_ERROR)
					XE_FAIL("Socket error");
				else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
				{
					// \note Socket should not be in non-blocking mode.
					DE_ASSERT(numRecv == 0);
					deYield();
				}
				else
				{
					DE_ASSERT(result == DE_SOCKETRESULT_SUCCESS);
					DE_ASSERT(numRecv <= bytesToRecv);
					m_curMsgPos += numRecv;
					// Continue receiving bytes / handle message in next iter.
				}
			}
		}
	}
	catch (const std::exception& e)
	{
		m_state.setState(COMMLINKSTATE_ERROR, e.what());
	}
}

void TcpIpRecvThread::stop (void)
{
	if (m_isRunning)
	{
		// \note Socket must be closed before terminating receive thread.
		XE_CHECK(!m_socket.isReceiveOpen());

		join();
		m_isRunning = false;
	}
}

void TcpIpRecvThread::handleMessage (xs::MessageType messageType, const deUint8* data, size_t dataSize)
{
	switch (messageType)
	{
		case xs::MESSAGETYPE_KEEPALIVE:
			m_state.onKeepaliveReceived();
			break;

		case xs::MESSAGETYPE_PROCESS_STARTED:
			XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_LAUNCHING, "Unexpected PROCESS_STARTED message");
			m_state.setState(COMMLINKSTATE_TEST_PROCESS_RUNNING);
			break;

		case xs::MESSAGETYPE_PROCESS_LAUNCH_FAILED:
		{
			xs::ProcessLaunchFailedMessage msg(data, dataSize);
			XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_LAUNCHING, "Unexpected PROCESS_LAUNCH_FAILED message");
			m_state.setState(COMMLINKSTATE_TEST_PROCESS_LAUNCH_FAILED, msg.reason.c_str());
			break;
		}

		case xs::MESSAGETYPE_PROCESS_FINISHED:
		{
			XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_RUNNING, "Unexpected PROCESS_FINISHED message");
			xs::ProcessFinishedMessage msg(data, dataSize);
			m_state.setState(COMMLINKSTATE_TEST_PROCESS_FINISHED);
			DE_UNREF(msg); // \todo [2012-06-19 pyry] Report exit code.
			break;
		}

		case xs::MESSAGETYPE_PROCESS_LOG_DATA:
		case xs::MESSAGETYPE_INFO:
			// Ignore leading \0 if such is present. \todo [2012-06-19 pyry] Improve protocol.
			if (data[dataSize-1] == 0)
				dataSize -= 1;

			if (messageType == xs::MESSAGETYPE_PROCESS_LOG_DATA)
			{
				XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_RUNNING, "Unexpected PROCESS_LOG_DATA message");
				m_state.onTestLogData(&data[0], dataSize);
			}
			else
				m_state.onInfoLogData(&data[0], dataSize);
			break;

		default:
			XE_FAIL("Unknown message");
	}
}

// TcpIpLink

TcpIpLink::TcpIpLink (void)
	: m_state			(COMMLINKSTATE_ERROR, "Not connected")
	, m_sendThread		(m_socket, m_state)
	, m_recvThread		(m_socket, m_state)
	, m_keepaliveTimer	(DE_NULL)
{
	m_keepaliveTimer = deTimer_create(keepaliveTimerCallback, this);
	XE_CHECK(m_keepaliveTimer);
}

TcpIpLink::~TcpIpLink (void)
{
	try
	{
		closeConnection();
	}
	catch (...)
	{
		// Can't do much except to ignore error.
	}
	deTimer_destroy(m_keepaliveTimer);
}

void TcpIpLink::closeConnection (void)
{
	{
		deSocketState state = m_socket.getState();
		if (state != DE_SOCKETSTATE_DISCONNECTED && state != DE_SOCKETSTATE_CLOSED)
			m_socket.shutdown();
	}

	if (deTimer_isActive(m_keepaliveTimer))
		deTimer_disable(m_keepaliveTimer);

	if (m_sendThread.isRunning())
		m_sendThread.stop();

	if (m_recvThread.isRunning())
		m_recvThread.stop();

	if (m_socket.getState() != DE_SOCKETSTATE_CLOSED)
		m_socket.close();
}

void TcpIpLink::connect (const de::SocketAddress& address)
{
	XE_CHECK(m_socket.getState() == DE_SOCKETSTATE_CLOSED);
	XE_CHECK(m_state.getState() == COMMLINKSTATE_ERROR);
	XE_CHECK(!m_sendThread.isRunning());
	XE_CHECK(!m_recvThread.isRunning());

	m_socket.connect(address);

	try
	{
		// Clear error and set state to ready.
		m_state.setState(COMMLINKSTATE_READY, "");
		m_state.onKeepaliveReceived();

		// Launch threads.
		m_sendThread.start();
		m_recvThread.start();

		XE_CHECK(deTimer_scheduleInterval(m_keepaliveTimer, xs::KEEPALIVE_SEND_INTERVAL));
	}
	catch (const std::exception& e)
	{
		closeConnection();
		m_state.setState(COMMLINKSTATE_ERROR, e.what());
		throw;
	}
}

void TcpIpLink::disconnect (void)
{
	try
	{
		closeConnection();
		m_state.setState(COMMLINKSTATE_ERROR, "Not connected");
	}
	catch (const std::exception& e)
	{
		m_state.setState(COMMLINKSTATE_ERROR, e.what());
	}
}

void TcpIpLink::reset (void)
{
	// \note Just clears error state if we are connected.
	if (m_socket.getState() == DE_SOCKETSTATE_CONNECTED)
	{
		m_state.setState(COMMLINKSTATE_READY, "");

		// \todo [2012-07-10 pyry] Do we need to reset send/receive buffers?
	}
	else
		disconnect(); // Abnormal state/usage. Disconnect socket.
}

void TcpIpLink::keepaliveTimerCallback (void* ptr)
{
	TcpIpLink*	link			= static_cast<TcpIpLink*>(ptr);
	deUint64	lastKeepalive	= link->m_state.getLastKeepaliveRecevied();
	deUint64	curTime			= deGetMicroseconds();

	// Check for timeout.
	if ((deInt64)curTime-(deInt64)lastKeepalive > xs::KEEPALIVE_TIMEOUT*1000)
		link->m_state.setState(COMMLINKSTATE_ERROR, "Keepalive timeout");

	// Enqueue new keepalive.
	try
	{
		writeKeepalive(link->m_sendThread.getBuffer());
	}
	catch (const de::BlockBuffer<deUint8>::CanceledException&)
	{
		// Ignore. Can happen in connection teardown.
	}
}

CommLinkState TcpIpLink::getState (void) const
{
	return m_state.getState();
}

CommLinkState TcpIpLink::getState (std::string& message) const
{
	return m_state.getState(message);
}

void TcpIpLink::setCallbacks (StateChangedFunc stateChangedCallback, LogDataFunc testLogDataCallback, LogDataFunc infoLogDataCallback, void* userPtr)
{
	m_state.setCallbacks(stateChangedCallback, testLogDataCallback, infoLogDataCallback, userPtr);
}

void TcpIpLink::startTestProcess (const char* name, const char* params, const char* workingDir, const char* caseList)
{
	XE_CHECK(m_state.getState() == COMMLINKSTATE_READY);

	m_state.setState(COMMLINKSTATE_TEST_PROCESS_LAUNCHING);
	writeExecuteBinary(m_sendThread.getBuffer(), name, params, workingDir, caseList);
}

void TcpIpLink::stopTestProcess (void)
{
	XE_CHECK(m_state.getState() != COMMLINKSTATE_ERROR);
	writeStopExecution(m_sendThread.getBuffer());
}

} // xe