1 /*------------------------------------------------------------------------- 2 * drawElements Quality Program Test Executor 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 Tcp/Ip communication link. 22 *//*--------------------------------------------------------------------*/ 23 24 #include "xeTcpIpLink.hpp" 25 #include "xsProtocol.hpp" 26 #include "deClock.h" 27 #include "deInt32.h" 28 29 namespace xe 30 { 31 32 enum 33 { 34 SEND_BUFFER_BLOCK_SIZE = 1024, 35 SEND_BUFFER_NUM_BLOCKS = 64 36 }; 37 38 // Utilities for writing messages out. 39 40 static void writeMessageHeader (de::BlockBuffer<deUint8>& dst, xs::MessageType type, int messageSize) 41 { 42 deUint8 hdr[xs::MESSAGE_HEADER_SIZE]; 43 xs::Message::writeHeader(type, messageSize, &hdr[0], xs::MESSAGE_HEADER_SIZE); 44 dst.write(xs::MESSAGE_HEADER_SIZE, &hdr[0]); 45 } 46 47 static void writeKeepalive (de::BlockBuffer<deUint8>& dst) 48 { 49 writeMessageHeader(dst, xs::MESSAGETYPE_KEEPALIVE, xs::MESSAGE_HEADER_SIZE); 50 dst.flush(); 51 } 52 53 static void writeExecuteBinary (de::BlockBuffer<deUint8>& dst, const char* name, const char* params, const char* workDir, const char* caseList) 54 { 55 int nameSize = (int)strlen(name) + 1; 56 int paramsSize = (int)strlen(params) + 1; 57 int workDirSize = (int)strlen(workDir) + 1; 58 int caseListSize = (int)strlen(caseList) + 1; 59 int totalSize = xs::MESSAGE_HEADER_SIZE + nameSize + paramsSize + workDirSize + caseListSize; 60 61 writeMessageHeader(dst, xs::MESSAGETYPE_EXECUTE_BINARY, totalSize); 62 dst.write(nameSize, (const deUint8*)name); 63 dst.write(paramsSize, (const deUint8*)params); 64 dst.write(workDirSize, (const deUint8*)workDir); 65 dst.write(caseListSize, (const deUint8*)caseList); 66 dst.flush(); 67 } 68 69 static void writeStopExecution (de::BlockBuffer<deUint8>& dst) 70 { 71 writeMessageHeader(dst, xs::MESSAGETYPE_STOP_EXECUTION, xs::MESSAGE_HEADER_SIZE); 72 dst.flush(); 73 } 74 75 // TcpIpLinkState 76 77 TcpIpLinkState::TcpIpLinkState (CommLinkState initialState, const char* initialErr) 78 : m_state (initialState) 79 , m_error (initialErr) 80 , m_lastKeepaliveReceived (0) 81 , m_stateChangedCallback (DE_NULL) 82 , m_testLogDataCallback (DE_NULL) 83 , m_infoLogDataCallback (DE_NULL) 84 , m_userPtr (DE_NULL) 85 { 86 } 87 88 TcpIpLinkState::~TcpIpLinkState (void) 89 { 90 } 91 92 CommLinkState TcpIpLinkState::getState (void) const 93 { 94 de::ScopedLock lock(m_lock); 95 96 return m_state; 97 } 98 99 CommLinkState TcpIpLinkState::getState (std::string& error) const 100 { 101 de::ScopedLock lock(m_lock); 102 103 error = m_error; 104 return m_state; 105 } 106 107 void TcpIpLinkState::setCallbacks (CommLink::StateChangedFunc stateChangedCallback, CommLink::LogDataFunc testLogDataCallback, CommLink::LogDataFunc infoLogDataCallback, void* userPtr) 108 { 109 de::ScopedLock lock(m_lock); 110 111 m_stateChangedCallback = stateChangedCallback; 112 m_testLogDataCallback = testLogDataCallback; 113 m_infoLogDataCallback = infoLogDataCallback; 114 m_userPtr = userPtr; 115 } 116 117 void TcpIpLinkState::setState (CommLinkState state, const char* error) 118 { 119 CommLink::StateChangedFunc callback = DE_NULL; 120 void* userPtr = DE_NULL; 121 122 { 123 de::ScopedLock lock(m_lock); 124 125 m_state = state; 126 m_error = error; 127 128 callback = m_stateChangedCallback; 129 userPtr = m_userPtr; 130 } 131 132 if (callback) 133 callback(userPtr, state, error); 134 } 135 136 void TcpIpLinkState::onTestLogData (const deUint8* bytes, size_t numBytes) const 137 { 138 CommLink::LogDataFunc callback = DE_NULL; 139 void* userPtr = DE_NULL; 140 141 m_lock.lock(); 142 callback = m_testLogDataCallback; 143 userPtr = m_userPtr; 144 m_lock.unlock(); 145 146 if (callback) 147 callback(userPtr, bytes, numBytes); 148 } 149 150 void TcpIpLinkState::onInfoLogData (const deUint8* bytes, size_t numBytes) const 151 { 152 CommLink::LogDataFunc callback = DE_NULL; 153 void* userPtr = DE_NULL; 154 155 m_lock.lock(); 156 callback = m_infoLogDataCallback; 157 userPtr = m_userPtr; 158 m_lock.unlock(); 159 160 if (callback) 161 callback(userPtr, bytes, numBytes); 162 } 163 164 void TcpIpLinkState::onKeepaliveReceived (void) 165 { 166 de::ScopedLock lock(m_lock); 167 m_lastKeepaliveReceived = deGetMicroseconds(); 168 } 169 170 deUint64 TcpIpLinkState::getLastKeepaliveRecevied (void) const 171 { 172 de::ScopedLock lock(m_lock); 173 return m_lastKeepaliveReceived; 174 } 175 176 // TcpIpSendThread 177 178 TcpIpSendThread::TcpIpSendThread (de::Socket& socket, TcpIpLinkState& state) 179 : m_socket (socket) 180 , m_state (state) 181 , m_buffer (SEND_BUFFER_BLOCK_SIZE, SEND_BUFFER_NUM_BLOCKS) 182 , m_isRunning (false) 183 { 184 } 185 186 TcpIpSendThread::~TcpIpSendThread (void) 187 { 188 } 189 190 void TcpIpSendThread::start (void) 191 { 192 DE_ASSERT(!m_isRunning); 193 194 // Reset state. 195 m_buffer.clear(); 196 m_isRunning = true; 197 198 de::Thread::start(); 199 } 200 201 void TcpIpSendThread::run (void) 202 { 203 try 204 { 205 deUint8 buf[SEND_BUFFER_BLOCK_SIZE]; 206 207 while (!m_buffer.isCanceled()) 208 { 209 size_t numToSend = 0; 210 size_t numSent = 0; 211 deSocketResult result = DE_SOCKETRESULT_LAST; 212 213 try 214 { 215 // Wait for single byte and then try to read more. 216 m_buffer.read(1, &buf[0]); 217 numToSend = 1 + m_buffer.tryRead(DE_LENGTH_OF_ARRAY(buf)-1, &buf[1]); 218 } 219 catch (const de::BlockBuffer<deUint8>::CanceledException&) 220 { 221 // Handled in loop condition. 222 } 223 224 while (numSent < numToSend) 225 { 226 result = m_socket.send(&buf[numSent], numToSend-numSent, &numSent); 227 228 if (result == DE_SOCKETRESULT_CONNECTION_CLOSED) 229 XE_FAIL("Connection closed"); 230 else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED) 231 XE_FAIL("Connection terminated"); 232 else if (result == DE_SOCKETRESULT_ERROR) 233 XE_FAIL("Socket error"); 234 else if (result == DE_SOCKETRESULT_WOULD_BLOCK) 235 { 236 // \note Socket should not be in non-blocking mode. 237 DE_ASSERT(numSent == 0); 238 deYield(); 239 } 240 else 241 DE_ASSERT(result == DE_SOCKETRESULT_SUCCESS); 242 } 243 } 244 } 245 catch (const std::exception& e) 246 { 247 m_state.setState(COMMLINKSTATE_ERROR, e.what()); 248 } 249 } 250 251 void TcpIpSendThread::stop (void) 252 { 253 if (m_isRunning) 254 { 255 m_buffer.cancel(); 256 join(); 257 m_isRunning = false; 258 } 259 } 260 261 // TcpIpRecvThread 262 263 TcpIpRecvThread::TcpIpRecvThread (de::Socket& socket, TcpIpLinkState& state) 264 : m_socket (socket) 265 , m_state (state) 266 , m_curMsgPos (0) 267 , m_isRunning (false) 268 { 269 } 270 271 TcpIpRecvThread::~TcpIpRecvThread (void) 272 { 273 } 274 275 void TcpIpRecvThread::start (void) 276 { 277 DE_ASSERT(!m_isRunning); 278 279 // Reset state. 280 m_curMsgPos = 0; 281 m_isRunning = true; 282 283 de::Thread::start(); 284 } 285 286 void TcpIpRecvThread::run (void) 287 { 288 try 289 { 290 for (;;) 291 { 292 bool hasHeader = m_curMsgPos >= xs::MESSAGE_HEADER_SIZE; 293 bool hasPayload = false; 294 size_t messageSize = 0; 295 xs::MessageType messageType = (xs::MessageType)0; 296 297 if (hasHeader) 298 { 299 xs::Message::parseHeader(&m_curMsgBuf[0], xs::MESSAGE_HEADER_SIZE, messageType, messageSize); 300 hasPayload = m_curMsgPos >= messageSize; 301 } 302 303 if (hasPayload) 304 { 305 // Process message. 306 handleMessage(messageType, m_curMsgPos > xs::MESSAGE_HEADER_SIZE ? &m_curMsgBuf[xs::MESSAGE_HEADER_SIZE] : DE_NULL, messageSize-xs::MESSAGE_HEADER_SIZE); 307 m_curMsgPos = 0; 308 } 309 else 310 { 311 // Try to receive missing bytes. 312 size_t curSize = hasHeader ? messageSize : (size_t)xs::MESSAGE_HEADER_SIZE; 313 size_t bytesToRecv = curSize-m_curMsgPos; 314 size_t numRecv = 0; 315 deSocketResult result = DE_SOCKETRESULT_LAST; 316 317 if (m_curMsgBuf.size() < curSize) 318 m_curMsgBuf.resize(curSize); 319 320 result = m_socket.receive(&m_curMsgBuf[m_curMsgPos], bytesToRecv, &numRecv); 321 322 if (result == DE_SOCKETRESULT_CONNECTION_CLOSED) 323 XE_FAIL("Connection closed"); 324 else if (result == DE_SOCKETRESULT_CONNECTION_TERMINATED) 325 XE_FAIL("Connection terminated"); 326 else if (result == DE_SOCKETRESULT_ERROR) 327 XE_FAIL("Socket error"); 328 else if (result == DE_SOCKETRESULT_WOULD_BLOCK) 329 { 330 // \note Socket should not be in non-blocking mode. 331 DE_ASSERT(numRecv == 0); 332 deYield(); 333 } 334 else 335 { 336 DE_ASSERT(result == DE_SOCKETRESULT_SUCCESS); 337 DE_ASSERT(numRecv <= bytesToRecv); 338 m_curMsgPos += numRecv; 339 // Continue receiving bytes / handle message in next iter. 340 } 341 } 342 } 343 } 344 catch (const std::exception& e) 345 { 346 m_state.setState(COMMLINKSTATE_ERROR, e.what()); 347 } 348 } 349 350 void TcpIpRecvThread::stop (void) 351 { 352 if (m_isRunning) 353 { 354 // \note Socket must be closed before terminating receive thread. 355 XE_CHECK(!m_socket.isReceiveOpen()); 356 357 join(); 358 m_isRunning = false; 359 } 360 } 361 362 void TcpIpRecvThread::handleMessage (xs::MessageType messageType, const deUint8* data, size_t dataSize) 363 { 364 switch (messageType) 365 { 366 case xs::MESSAGETYPE_KEEPALIVE: 367 m_state.onKeepaliveReceived(); 368 break; 369 370 case xs::MESSAGETYPE_PROCESS_STARTED: 371 XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_LAUNCHING, "Unexpected PROCESS_STARTED message"); 372 m_state.setState(COMMLINKSTATE_TEST_PROCESS_RUNNING); 373 break; 374 375 case xs::MESSAGETYPE_PROCESS_LAUNCH_FAILED: 376 { 377 xs::ProcessLaunchFailedMessage msg(data, dataSize); 378 XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_LAUNCHING, "Unexpected PROCESS_LAUNCH_FAILED message"); 379 m_state.setState(COMMLINKSTATE_TEST_PROCESS_LAUNCH_FAILED, msg.reason.c_str()); 380 break; 381 } 382 383 case xs::MESSAGETYPE_PROCESS_FINISHED: 384 { 385 XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_RUNNING, "Unexpected PROCESS_FINISHED message"); 386 xs::ProcessFinishedMessage msg(data, dataSize); 387 m_state.setState(COMMLINKSTATE_TEST_PROCESS_FINISHED); 388 DE_UNREF(msg); // \todo [2012-06-19 pyry] Report exit code. 389 break; 390 } 391 392 case xs::MESSAGETYPE_PROCESS_LOG_DATA: 393 case xs::MESSAGETYPE_INFO: 394 // Ignore leading \0 if such is present. \todo [2012-06-19 pyry] Improve protocol. 395 if (data[dataSize-1] == 0) 396 dataSize -= 1; 397 398 if (messageType == xs::MESSAGETYPE_PROCESS_LOG_DATA) 399 { 400 XE_CHECK_MSG(m_state.getState() == COMMLINKSTATE_TEST_PROCESS_RUNNING, "Unexpected PROCESS_LOG_DATA message"); 401 m_state.onTestLogData(&data[0], dataSize); 402 } 403 else 404 m_state.onInfoLogData(&data[0], dataSize); 405 break; 406 407 default: 408 XE_FAIL("Unknown message"); 409 } 410 } 411 412 // TcpIpLink 413 414 TcpIpLink::TcpIpLink (void) 415 : m_state (COMMLINKSTATE_ERROR, "Not connected") 416 , m_sendThread (m_socket, m_state) 417 , m_recvThread (m_socket, m_state) 418 , m_keepaliveTimer (DE_NULL) 419 { 420 m_keepaliveTimer = deTimer_create(keepaliveTimerCallback, this); 421 XE_CHECK(m_keepaliveTimer); 422 } 423 424 TcpIpLink::~TcpIpLink (void) 425 { 426 try 427 { 428 closeConnection(); 429 } 430 catch (...) 431 { 432 // Can't do much except to ignore error. 433 } 434 deTimer_destroy(m_keepaliveTimer); 435 } 436 437 void TcpIpLink::closeConnection (void) 438 { 439 { 440 deSocketState state = m_socket.getState(); 441 if (state != DE_SOCKETSTATE_DISCONNECTED && state != DE_SOCKETSTATE_CLOSED) 442 m_socket.shutdown(); 443 } 444 445 if (deTimer_isActive(m_keepaliveTimer)) 446 deTimer_disable(m_keepaliveTimer); 447 448 if (m_sendThread.isRunning()) 449 m_sendThread.stop(); 450 451 if (m_recvThread.isRunning()) 452 m_recvThread.stop(); 453 454 if (m_socket.getState() != DE_SOCKETSTATE_CLOSED) 455 m_socket.close(); 456 } 457 458 void TcpIpLink::connect (const de::SocketAddress& address) 459 { 460 XE_CHECK(m_socket.getState() == DE_SOCKETSTATE_CLOSED); 461 XE_CHECK(m_state.getState() == COMMLINKSTATE_ERROR); 462 XE_CHECK(!m_sendThread.isRunning()); 463 XE_CHECK(!m_recvThread.isRunning()); 464 465 m_socket.connect(address); 466 467 try 468 { 469 // Clear error and set state to ready. 470 m_state.setState(COMMLINKSTATE_READY, ""); 471 m_state.onKeepaliveReceived(); 472 473 // Launch threads. 474 m_sendThread.start(); 475 m_recvThread.start(); 476 477 XE_CHECK(deTimer_scheduleInterval(m_keepaliveTimer, xs::KEEPALIVE_SEND_INTERVAL)); 478 } 479 catch (const std::exception& e) 480 { 481 closeConnection(); 482 m_state.setState(COMMLINKSTATE_ERROR, e.what()); 483 throw; 484 } 485 } 486 487 void TcpIpLink::disconnect (void) 488 { 489 try 490 { 491 closeConnection(); 492 m_state.setState(COMMLINKSTATE_ERROR, "Not connected"); 493 } 494 catch (const std::exception& e) 495 { 496 m_state.setState(COMMLINKSTATE_ERROR, e.what()); 497 } 498 } 499 500 void TcpIpLink::reset (void) 501 { 502 // \note Just clears error state if we are connected. 503 if (m_socket.getState() == DE_SOCKETSTATE_CONNECTED) 504 { 505 m_state.setState(COMMLINKSTATE_READY, ""); 506 507 // \todo [2012-07-10 pyry] Do we need to reset send/receive buffers? 508 } 509 else 510 disconnect(); // Abnormal state/usage. Disconnect socket. 511 } 512 513 void TcpIpLink::keepaliveTimerCallback (void* ptr) 514 { 515 TcpIpLink* link = static_cast<TcpIpLink*>(ptr); 516 deUint64 lastKeepalive = link->m_state.getLastKeepaliveRecevied(); 517 deUint64 curTime = deGetMicroseconds(); 518 519 // Check for timeout. 520 if ((deInt64)curTime-(deInt64)lastKeepalive > xs::KEEPALIVE_TIMEOUT*1000) 521 link->m_state.setState(COMMLINKSTATE_ERROR, "Keepalive timeout"); 522 523 // Enqueue new keepalive. 524 try 525 { 526 writeKeepalive(link->m_sendThread.getBuffer()); 527 } 528 catch (const de::BlockBuffer<deUint8>::CanceledException&) 529 { 530 // Ignore. Can happen in connection teardown. 531 } 532 } 533 534 CommLinkState TcpIpLink::getState (void) const 535 { 536 return m_state.getState(); 537 } 538 539 CommLinkState TcpIpLink::getState (std::string& message) const 540 { 541 return m_state.getState(message); 542 } 543 544 void TcpIpLink::setCallbacks (StateChangedFunc stateChangedCallback, LogDataFunc testLogDataCallback, LogDataFunc infoLogDataCallback, void* userPtr) 545 { 546 m_state.setCallbacks(stateChangedCallback, testLogDataCallback, infoLogDataCallback, userPtr); 547 } 548 549 void TcpIpLink::startTestProcess (const char* name, const char* params, const char* workingDir, const char* caseList) 550 { 551 XE_CHECK(m_state.getState() == COMMLINKSTATE_READY); 552 553 m_state.setState(COMMLINKSTATE_TEST_PROCESS_LAUNCHING); 554 writeExecuteBinary(m_sendThread.getBuffer(), name, params, workingDir, caseList); 555 } 556 557 void TcpIpLink::stopTestProcess (void) 558 { 559 XE_CHECK(m_state.getState() != COMMLINKSTATE_ERROR); 560 writeStopExecution(m_sendThread.getBuffer()); 561 } 562 563 } // xe 564
