execserver/xsExecutionServer.cpp

   1 /*-------------------------------------------------------------------------
   2  * drawElements Quality Program Execution Server
   3  * ---------------------------------------------
   4  *
   5  * Copyright 2014 The Android Open Source Project
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  *//*!
  20  * \file
  21  * \brief Test Execution Server.
  22  *//*--------------------------------------------------------------------*/
  23
  24 #include "xsExecutionServer.hpp"
  25 #include "deClock.h"
  26
  27 #include <cstdio>
  28
  29 using std::vector;
  30 using std::string;
  31
  32 #if 1
  33 #       define DBG_PRINT(X) printf X
  34 #else
  35 #       define DBG_PRINT(X)
  36 #endif
  37
  38 namespace xs
  39 {
  40
  41 inline bool MessageBuilder::isComplete (void) const
  42 {
  43         if (m_buffer.size() < MESSAGE_HEADER_SIZE)
  44                 return false;
  45         else
  46                 return m_buffer.size() == getMessageSize();
  47 }
  48
  49 const deUint8* MessageBuilder::getMessageData (void) const
  50 {
  51         return m_buffer.size() > MESSAGE_HEADER_SIZE ? &m_buffer[MESSAGE_HEADER_SIZE] : DE_NULL;
  52 }
  53
  54 size_t MessageBuilder::getMessageDataSize (void) const
  55 {
  56         DE_ASSERT(isComplete());
  57         return m_buffer.size() - MESSAGE_HEADER_SIZE;
  58 }
  59
  60 void MessageBuilder::read (ByteBuffer& src)
  61 {
  62         // Try to get header.
  63         if (m_buffer.size() < MESSAGE_HEADER_SIZE)
  64         {
  65                 while (m_buffer.size() < MESSAGE_HEADER_SIZE &&
  66                            src.getNumElements() > 0)
  67                         m_buffer.push_back(src.popBack());
  68
  69                 DE_ASSERT(m_buffer.size() <= MESSAGE_HEADER_SIZE);
  70
  71                 if (m_buffer.size() == MESSAGE_HEADER_SIZE)
  72                 {
  73                         // Got whole header, parse it.
  74                         Message::parseHeader(&m_buffer[0], (int)m_buffer.size(), m_messageType, m_messageSize);
  75                 }
  76         }
  77
  78         if (m_buffer.size() >= MESSAGE_HEADER_SIZE)
  79         {
  80                 // We have header.
  81                 size_t msgSize                  = getMessageSize();
  82                 size_t numBytesLeft             = msgSize - m_buffer.size();
  83                 size_t numToRead                = (size_t)de::min(src.getNumElements(), (int)numBytesLeft);
  84
  85                 if (numToRead > 0)
  86                 {
  87                         int curBufPos = (int)m_buffer.size();
  88                         m_buffer.resize(curBufPos+numToRead);
  89                         src.popBack(&m_buffer[curBufPos], (int)numToRead);
  90                 }
  91         }
  92 }
  93
  94 void MessageBuilder::clear (void)
  95 {
  96         m_buffer.clear();
  97         m_messageType   = MESSAGETYPE_NONE;
  98         m_messageSize   = 0;
  99 }
 100
 101 ExecutionServer::ExecutionServer (xs::TestProcess* testProcess, deSocketFamily family, int port, RunMode runMode)
 102         : TcpServer             (family, port)
 103         , m_testDriver  (testProcess)
 104         , m_runMode             (runMode)
 105 {
 106 }
 107
 108 ExecutionServer::~ExecutionServer (void)
 109 {
 110 }
 111
 112 TestDriver* ExecutionServer::acquireTestDriver (void)
 113 {
 114         if (!m_testDriverLock.tryLock())
 115                 throw Error("Failed to acquire test driver");
 116
 117         return &m_testDriver;
 118 }
 119
 120 void ExecutionServer::releaseTestDriver (TestDriver* driver)
 121 {
 122         DE_ASSERT(&m_testDriver == driver);
 123         DE_UNREF(driver);
 124         m_testDriverLock.unlock();
 125 }
 126
 127 ConnectionHandler* ExecutionServer::createHandler (de::Socket* socket, const de::SocketAddress& clientAddress)
 128 {
 129         printf("ExecutionServer: New connection from %s:%d\n", clientAddress.getHost(), clientAddress.getPort());
 130         return new ExecutionRequestHandler(this, socket);
 131 }
 132
 133 void ExecutionServer::connectionDone (ConnectionHandler* handler)
 134 {
 135         if (m_runMode == RUNMODE_SINGLE_EXEC)
 136                 m_socket.close();
 137
 138         TcpServer::connectionDone(handler);
 139 }
 140
 141 ExecutionRequestHandler::ExecutionRequestHandler (ExecutionServer* server, de::Socket* socket)
 142         : ConnectionHandler     (server, socket)
 143         , m_execServer          (server)
 144         , m_testDriver          (DE_NULL)
 145         , m_bufferIn            (RECV_BUFFER_SIZE)
 146         , m_bufferOut           (SEND_BUFFER_SIZE)
 147         , m_run                         (false)
 148         , m_sendRecvTmpBuf      (SEND_RECV_TMP_BUFFER_SIZE)
 149 {
 150         // Set flags.
 151         m_socket->setFlags(DE_SOCKET_NONBLOCKING|DE_SOCKET_KEEPALIVE|DE_SOCKET_CLOSE_ON_EXEC);
 152
 153         // Init protocol keepalives.
 154         initKeepAlives();
 155 }
 156
 157 ExecutionRequestHandler::~ExecutionRequestHandler (void)
 158 {
 159         if (m_testDriver)
 160                 m_execServer->releaseTestDriver(m_testDriver);
 161 }
 162
 163 void ExecutionRequestHandler::handle (void)
 164 {
 165         DBG_PRINT(("ExecutionRequestHandler::handle()\n"));
 166
 167         try
 168         {
 169                 // Process execution session.
 170                 processSession();
 171         }
 172         catch (const std::exception& e)
 173         {
 174                 printf("ExecutionRequestHandler::run(): %s\n", e.what());
 175         }
 176
 177         DBG_PRINT(("ExecutionRequestHandler::handle(): Done!\n"));
 178
 179         // Release test driver.
 180         if (m_testDriver)
 181         {
 182                 try
 183                 {
 184                         m_testDriver->reset();
 185                 }
 186                 catch (...)
 187                 {
 188                 }
 189                 m_execServer->releaseTestDriver(m_testDriver);
 190                 m_testDriver = DE_NULL;
 191         }
 192
 193         // Close connection.
 194         if (m_socket->isConnected())
 195                 m_socket->shutdown();
 196 }
 197
 198 void ExecutionRequestHandler::acquireTestDriver (void)
 199 {
 200         DE_ASSERT(!m_testDriver);
 201
 202         // Try to acquire test driver - may fail.
 203         m_testDriver = m_execServer->acquireTestDriver();
 204         DE_ASSERT(m_testDriver);
 205         m_testDriver->reset();
 206
 207 }
 208
 209 void ExecutionRequestHandler::processSession (void)
 210 {
 211         m_run = true;
 212
 213         deUint64 lastIoTime = deGetMicroseconds();
 214
 215         while (m_run)
 216         {
 217                 bool anyIO = false;
 218
 219                 // Read from socket to buffer.
 220                 anyIO = receive() || anyIO;
 221
 222                 // Send bytes in buffer.
 223                 anyIO = send() || anyIO;
 224
 225                 // Process incoming data.
 226                 if (m_bufferIn.getNumElements() > 0)
 227                 {
 228                         DE_ASSERT(!m_msgBuilder.isComplete());
 229                         m_msgBuilder.read(m_bufferIn);
 230                 }
 231
 232                 if (m_msgBuilder.isComplete())
 233                 {
 234                         // Process message.
 235                         processMessage(m_msgBuilder.getMessageType(), m_msgBuilder.getMessageData(), m_msgBuilder.getMessageDataSize());
 236
 237                         m_msgBuilder.clear();
 238                 }
 239
 240                 // Keepalives, anyone?
 241                 pollKeepAlives();
 242
 243                 // Poll test driver for IO.
 244                 if (m_testDriver)
 245                         anyIO = getTestDriver()->poll(m_bufferOut) || anyIO;
 246
 247                 // If no IO happens in a reasonable amount of time, go to sleep.
 248                 {
 249                         deUint64 curTime = deGetMicroseconds();
 250                         if (anyIO)
 251                                 lastIoTime = curTime;
 252                         else if (curTime-lastIoTime > SERVER_IDLE_THRESHOLD*1000)
 253                                 deSleep(SERVER_IDLE_SLEEP); // Too long since last IO, sleep for a while.
 254                         else
 255                                 deYield(); // Just give other threads chance to run.
 256                 }
 257         }
 258 }
 259
 260 void ExecutionRequestHandler::processMessage (MessageType type, const deUint8* data, size_t dataSize)
 261 {
 262         switch (type)
 263         {
 264                 case MESSAGETYPE_HELLO:
 265                 {
 266                         HelloMessage msg(data, dataSize);
 267                         DBG_PRINT(("HelloMessage: version = %d\n", msg.version));
 268                         if (msg.version != PROTOCOL_VERSION)
 269                                 throw ProtocolError("Unsupported protocol version");
 270                         break;
 271                 }
 272
 273                 case MESSAGETYPE_TEST:
 274                 {
 275                         TestMessage msg(data, dataSize);
 276                         DBG_PRINT(("TestMessage: '%s'\n", msg.test.c_str()));
 277                         break;
 278                 }
 279
 280                 case MESSAGETYPE_KEEPALIVE:
 281                 {
 282                         KeepAliveMessage msg(data, dataSize);
 283                         DBG_PRINT(("KeepAliveMessage\n"));
 284                         keepAliveReceived();
 285                         break;
 286                 }
 287
 288                 case MESSAGETYPE_EXECUTE_BINARY:
 289                 {
 290                         ExecuteBinaryMessage msg(data, dataSize);
 291                         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()));
 292                         getTestDriver()->startProcess(msg.name.c_str(), msg.params.c_str(), msg.workDir.c_str(), msg.caseList.c_str());
 293                         keepAliveReceived(); // \todo [2011-10-11 pyry] Remove this once Candy is fixed.
 294                         break;
 295                 }
 296
 297                 case MESSAGETYPE_STOP_EXECUTION:
 298                 {
 299                         StopExecutionMessage msg(data, dataSize);
 300                         DBG_PRINT(("StopExecutionMessage\n"));
 301                         getTestDriver()->stopProcess();
 302                         break;
 303                 }
 304
 305                 default:
 306                         throw ProtocolError("Unsupported message");
 307         }
 308 }
 309
 310 void ExecutionRequestHandler::initKeepAlives (void)
 311 {
 312         deUint64 curTime = deGetMicroseconds();
 313         m_lastKeepAliveSent             = curTime;
 314         m_lastKeepAliveReceived = curTime;
 315 }
 316
 317 void ExecutionRequestHandler::keepAliveReceived (void)
 318 {
 319         m_lastKeepAliveReceived = deGetMicroseconds();
 320 }
 321
 322 void ExecutionRequestHandler::pollKeepAlives (void)
 323 {
 324         deUint64 curTime = deGetMicroseconds();
 325
 326         // Check that we've got keepalives in timely fashion.
 327         if (curTime - m_lastKeepAliveReceived > KEEPALIVE_TIMEOUT*1000)
 328                 throw ProtocolError("Keepalive timeout occurred");
 329
 330         // Send some?
 331         if (curTime - m_lastKeepAliveSent > KEEPALIVE_SEND_INTERVAL*1000 &&
 332                 m_bufferOut.getNumFree() >= MESSAGE_HEADER_SIZE)
 333         {
 334                 vector<deUint8> buf;
 335                 KeepAliveMessage().write(buf);
 336                 m_bufferOut.pushFront(&buf[0], (int)buf.size());
 337
 338                 m_lastKeepAliveSent = deGetMicroseconds();
 339         }
 340 }
 341
 342 bool ExecutionRequestHandler::receive (void)
 343 {
 344         size_t maxLen = de::min(m_sendRecvTmpBuf.size(), (size_t)m_bufferIn.getNumFree());
 345
 346         if (maxLen > 0)
 347         {
 348                 size_t                  numRecv;
 349                 deSocketResult  result  = m_socket->receive(&m_sendRecvTmpBuf[0], maxLen, &numRecv);
 350
 351                 if (result == DE_SOCKETRESULT_SUCCESS)
 352                 {
 353                         DE_ASSERT(numRecv > 0);
 354                         m_bufferIn.pushFront(&m_sendRecvTmpBuf[0], (int)numRecv);
 355                         return true;
 356                 }
 357                 else if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
 358                 {
 359                         m_run = false;
 360                         return true;
 361                 }
 362                 else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
 363                         return false;
 364                 else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
 365                         throw ConnectionError("Connection terminated");
 366                 else
 367                         throw ConnectionError("receive() failed");
 368         }
 369         else
 370                 return false;
 371 }
 372
 373 bool ExecutionRequestHandler::send (void)
 374 {
 375         size_t maxLen = de::min(m_sendRecvTmpBuf.size(), (size_t)m_bufferOut.getNumElements());
 376
 377         if (maxLen > 0)
 378         {
 379                 m_bufferOut.peekBack(&m_sendRecvTmpBuf[0], (int)maxLen);
 380
 381                 size_t                  numSent;
 382                 deSocketResult  result  = m_socket->send(&m_sendRecvTmpBuf[0], maxLen, &numSent);
 383
 384                 if (result == DE_SOCKETRESULT_SUCCESS)
 385                 {
 386                         DE_ASSERT(numSent > 0);
 387                         m_bufferOut.popBack((int)numSent);
 388                         return true;
 389                 }
 390                 else if (result == DE_SOCKETRESULT_CONNECTION_CLOSED)
 391                 {
 392                         m_run = false;
 393                         return true;
 394                 }
 395                 else if (result == DE_SOCKETRESULT_WOULD_BLOCK)
 396                         return false;
 397                 else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED)
 398                         throw ConnectionError("Connection terminated");
 399                 else
 400                         throw ConnectionError("send() failed");
 401         }
 402         else
 403                 return false;
 404 }
 405
 406 } // xs