blob: d99c8058576a4d39b18c19b34a53ff06fb03da67 [file] [log] [blame]
/*-------------------------------------------------------------------------
* drawElements Quality Program Execution Server
* ---------------------------------------------
*
* 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 Test Execution Server.
*//*--------------------------------------------------------------------*/
#include "xsExecutionServer.hpp"
#include "deClock.h"
#include <cstdio>
using std::vector;
using std::string;
#if 1
# define DBG_PRINT(X) printf X
#else
# define DBG_PRINT(X)
#endif
namespace xs
{
inline bool MessageBuilder::isComplete (void) const
{
if (m_buffer.size() < MESSAGE_HEADER_SIZE)
return false;
else
return m_buffer.size() == getMessageSize();
}
const deUint8* MessageBuilder::getMessageData (void) const
{
return m_buffer.size() > MESSAGE_HEADER_SIZE ? &m_buffer[MESSAGE_HEADER_SIZE] : DE_NULL;
}
size_t MessageBuilder::getMessageDataSize (void) const
{
DE_ASSERT(isComplete());
return m_buffer.size() - MESSAGE_HEADER_SIZE;
}
void MessageBuilder::read (ByteBuffer& src)
{
// Try to get header.
if (m_buffer.size() < MESSAGE_HEADER_SIZE)
{
while (m_buffer.size() < MESSAGE_HEADER_SIZE &&
src.getNumElements() > 0)
m_buffer.push_back(src.popBack());
DE_ASSERT(m_buffer.size() <= MESSAGE_HEADER_SIZE);
if (m_buffer.size() == MESSAGE_HEADER_SIZE)
{
// Got whole header, parse it.
Message::parseHeader(&m_buffer[0], (int)m_buffer.size(), m_messageType, m_messageSize);
}
}
if (m_buffer.size() >= MESSAGE_HEADER_SIZE)
{
// We have header.
size_t msgSize = getMessageSize();
size_t numBytesLeft = msgSize - m_buffer.size();
size_t numToRead = (size_t)de::min(src.getNumElements(), (int)numBytesLeft);
if (numToRead > 0)
{
int curBufPos = (int)m_buffer.size();
m_buffer.resize(curBufPos+numToRead);
src.popBack(&m_buffer[curBufPos], (int)numToRead);
}
}
}
void MessageBuilder::clear (void)
{
m_buffer.clear();
m_messageType = MESSAGETYPE_NONE;
m_messageSize = 0;
}
ExecutionServer::ExecutionServer (xs::TestProcess* testProcess, deSocketFamily family, int port, RunMode runMode)
: TcpServer (family, port)
, m_testDriver (testProcess)
, m_runMode (runMode)
{
}
ExecutionServer::~ExecutionServer (void)
{
}
TestDriver* ExecutionServer::acquireTestDriver (void)
{
if (!m_testDriverLock.tryLock())
throw Error("Failed to acquire test driver");
return &m_testDriver;
}
void ExecutionServer::releaseTestDriver (TestDriver* driver)
{
DE_ASSERT(&m_testDriver == driver);
DE_UNREF(driver);
m_testDriverLock.unlock();
}
ConnectionHandler* ExecutionServer::createHandler (de::Socket* socket, const de::SocketAddress& clientAddress)
{
printf("ExecutionServer: New connection from %s:%d\n", clientAddress.getHost(), clientAddress.getPort());
return new ExecutionRequestHandler(this, socket);
}
void ExecutionServer::connectionDone (ConnectionHandler* handler)
{
if (m_runMode == RUNMODE_SINGLE_EXEC)
m_socket.close();
TcpServer::connectionDone(handler);
}
ExecutionRequestHandler::ExecutionRequestHandler (ExecutionServer* server, de::Socket* socket)
: ConnectionHandler (server, socket)
, m_execServer (server)
, m_testDriver (DE_NULL)
, m_bufferIn (RECV_BUFFER_SIZE)
, m_bufferOut (SEND_BUFFER_SIZE)
, m_run (false)
, m_sendRecvTmpBuf (SEND_RECV_TMP_BUFFER_SIZE)
{
// Set flags.
m_socket->setFlags(DE_SOCKET_NONBLOCKING|DE_SOCKET_KEEPALIVE|DE_SOCKET_CLOSE_ON_EXEC);
// Init protocol keepalives.
initKeepAlives();
}
ExecutionRequestHandler::~ExecutionRequestHandler (void)
{
if (m_testDriver)
m_execServer->releaseTestDriver(m_testDriver);
}
void ExecutionRequestHandler::handle (void)
{
DBG_PRINT(("ExecutionRequestHandler::handle()\n"));
try
{
// Process execution session.
processSession();
}
catch (const std::exception& e)
{
printf("ExecutionRequestHandler::run(): %s\n", e.what());
}
DBG_PRINT(("ExecutionRequestHandler::handle(): Done!\n"));
// Release test driver.
if (m_testDriver)
{
try
{
m_testDriver->reset();
}
catch (...)
{
}
m_execServer->releaseTestDriver(m_testDriver);
m_testDriver = DE_NULL;
}
// Close connection.
if (m_socket->isConnected())
m_socket->shutdown();
}
void ExecutionRequestHandler::acquireTestDriver (void)
{
DE_ASSERT(!m_testDriver);
// Try to acquire test driver - may fail.
m_testDriver = m_execServer->acquireTestDriver();
DE_ASSERT(m_testDriver);
m_testDriver->reset();
}
void ExecutionRequestHandler::processSession (void)
{
m_run = true;
deUint64 lastIoTime = deGetMicroseconds();
while (m_run)
{
bool anyIO = false;
// Read from socket to buffer.
anyIO = receive() || anyIO;
// Send bytes in buffer.
anyIO = send() || anyIO;
// Process incoming data.
if (m_bufferIn.getNumElements() > 0)
{
DE_ASSERT(!m_msgBuilder.isComplete());
m_msgBuilder.read(m_bufferIn);
}
if (m_msgBuilder.isComplete())
{
// Process message.
processMessage(m_msgBuilder.getMessageType(), m_msgBuilder.getMessageData(), m_msgBuilder.getMessageDataSize());
m_msgBuilder.clear();
}
// Keepalives, anyone?
pollKeepAlives();
// Poll test driver for IO.
if (m_testDriver)
anyIO = getTestDriver()->poll(m_bufferOut) || anyIO;
// If no IO happens in a reasonable amount of time, go to sleep.
{
deUint64 curTime = deGetMicroseconds();
if (anyIO)
lastIoTime = curTime;
else if (curTime-lastIoTime > SERVER_IDLE_THRESHOLD*1000)
deSleep(SERVER_IDLE_SLEEP); // Too long since last IO, sleep for a while.
else
deYield(); // Just give other threads chance to run.
}
}
}
void ExecutionRequestHandler::processMessage (MessageType type, const deUint8* data, size_t dataSize)
{
switch (type)
{
case MESSAGETYPE_HELLO:
{
HelloMessage msg(data, dataSize);
DBG_PRINT(("HelloMessage: version = %d\n", msg.version));
if (msg.version != PROTOCOL_VERSION)
throw ProtocolError("Unsupported protocol version");
break;
}
case MESSAGETYPE_TEST:
{
TestMessage msg(data, dataSize);
DBG_PRINT(("TestMessage: '%s'\n", msg.test.c_str()));
break;
}
case MESSAGETYPE_KEEPALIVE:
{
KeepAliveMessage msg(data, dataSize);
DBG_PRINT(("KeepAliveMessage\n"));
keepAliveReceived();
break;
}
case MESSAGETYPE_EXECUTE_BINARY:
{
ExecuteBinaryMessage msg(data, dataSize);
DBG_PRINT(("ExecuteBinaryMessage: '%s', '%s', '%s', '%s'\n", msg.name.c_str(), msg.params.c_str(), msg.workDir.c_str(), msg.caseList.substr(0, 10).c_str()));
getTestDriver()->startProcess(msg.name.c_str(), msg.params.c_str(), msg.workDir.c_str(), msg.caseList.c_str());
keepAliveReceived(); // \todo [2011-10-11 pyry] Remove this once Candy is fixed.
break;
}
case MESSAGETYPE_STOP_EXECUTION:
{
StopExecutionMessage msg(data, dataSize);
DBG_PRINT(("StopExecutionMessage\n"));
getTestDriver()->stopProcess();
break;
}
default:
throw ProtocolError("Unsupported message");
}
}
void ExecutionRequestHandler::initKeepAlives (void)
{
deUint64 curTime = deGetMicroseconds();
m_lastKeepAliveSent = curTime;
m_lastKeepAliveReceived = curTime;
}
void ExecutionRequestHandler::keepAliveReceived (void)
{
m_lastKeepAliveReceived = deGetMicroseconds();
}
void ExecutionRequestHandler::pollKeepAlives (void)
{
deUint64 curTime = deGetMicroseconds();
// Check that we've got keepalives in timely fashion.
if (curTime - m_lastKeepAliveReceived > KEEPALIVE_TIMEOUT*1000)
throw ProtocolError("Keepalive timeout occurred");
// Send some?
if (curTime - m_lastKeepAliveSent > KEEPALIVE_SEND_INTERVAL*1000 &&
m_bufferOut.getNumFree() >= MESSAGE_HEADER_SIZE)
{
vector<deUint8> buf;
KeepAliveMessage().write(buf);
m_bufferOut.pushFront(&buf[0], (int)buf.size());
m_lastKeepAliveSent = deGetMicroseconds();
}
}
bool ExecutionRequestHandler::receive (void)
{
size_t maxLen = de::min(m_sendRecvTmpBuf.size(), (size_t)m_bufferIn.getNumFree());
if (maxLen > 0)
{
size_t numRecv;
deSocketResult result = m_socket->receive(&m_sendRecvTmpBuf[0], maxLen, &numRecv);
if (result == DE_SOCKETRESULT_SUCCESS)
{
DE_ASSERT(numRecv > 0);
m_bufferIn.pushFront(&m_sendRecvTmpBuf[0], (int)numRecv);
return true;
}
else if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
{
m_run = false;
return true;
}
else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
return false;
else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
throw ConnectionError("Connection terminated");
else
throw ConnectionError("receive() failed");
}
else
return false;
}
bool ExecutionRequestHandler::send (void)
{
size_t maxLen = de::min(m_sendRecvTmpBuf.size(), (size_t)m_bufferOut.getNumElements());
if (maxLen > 0)
{
m_bufferOut.peekBack(&m_sendRecvTmpBuf[0], (int)maxLen);
size_t numSent;
deSocketResult result = m_socket->send(&m_sendRecvTmpBuf[0], maxLen, &numSent);
if (result == DE_SOCKETRESULT_SUCCESS)
{
DE_ASSERT(numSent > 0);
m_bufferOut.popBack((int)numSent);
return true;
}
else if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
{
m_run = false;
return true;
}
else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
return false;
else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
throw ConnectionError("Connection terminated");
else
throw ConnectionError("send() failed");
}
else
return false;
}
} // xs