| 69 | ~Task() override = default; |
| 70 | |
| 71 | void run() override { |
| 72 | shared_ptr<TServerEventHandler> eventHandler = server_.getEventHandler(); |
| 73 | void* connectionContext = nullptr; |
| 74 | if (eventHandler != nullptr) { |
| 75 | connectionContext = eventHandler->createContext(input_, output_); |
| 76 | } |
| 77 | try { |
| 78 | for (;;) { |
| 79 | if (eventHandler != nullptr) { |
| 80 | eventHandler->processContext(connectionContext, transport_); |
| 81 | } |
| 82 | // Setting a socket timeout for process() may lead to false positive |
| 83 | // and prematurely closes a slow client's connection. |
| 84 | if (!processor_->process(input_, output_, connectionContext) |
| 85 | || !Peek(connectionContext, eventHandler.get())) { |
| 86 | break; |
| 87 | } |
| 88 | } |
| 89 | } catch (const TTransportException& ttx) { |
| 90 | // IMPALA-13020: Thrift throws an END_OF_FILE exception when it hits the |
| 91 | // max message size. That is always interesting to us, so we specifically |
| 92 | // detect "MaxMessageSize" and print it along with advice on how to address it. |
| 93 | bool hit_max_message_size = |
| 94 | std::string(ttx.what()).find("MaxMessageSize") != std::string::npos; |
| 95 | if (ttx.getType() != TTransportException::END_OF_FILE || hit_max_message_size) { |
| 96 | string errStr = string("TAcceptQueueServer client died: ") + ttx.what(); |
| 97 | GlobalOutput(errStr.c_str()); |
| 98 | if (hit_max_message_size) { |
| 99 | GlobalOutput("MaxMessageSize errors can be addressed by increasing " |
| 100 | "thrift_rpc_max_message_size on the receiving nodes."); |
| 101 | } |
| 102 | } |
| 103 | } catch (const std::exception& x) { |
| 104 | GlobalOutput.printf( |
| 105 | "TAcceptQueueServer exception: %s: %s", typeid(x).name(), x.what()); |
| 106 | } catch (...) { |
| 107 | GlobalOutput("TAcceptQueueServer uncaught exception."); |
| 108 | } |
| 109 | |
| 110 | try { |
| 111 | input_->getTransport()->close(); |
| 112 | } catch (const TTransportException& ttx) { |
| 113 | string errStr = string("TAcceptQueueServer input close failed: ") + ttx.what(); |
| 114 | GlobalOutput(errStr.c_str()); |
| 115 | } |
| 116 | try { |
| 117 | output_->getTransport()->close(); |
| 118 | } catch (const TTransportException& ttx) { |
| 119 | string errStr = string("TAcceptQueueServer output close failed: ") + ttx.what(); |
| 120 | GlobalOutput(errStr.c_str()); |
| 121 | } |
| 122 | |
| 123 | // Delete the context after closing the transports in case they have references to it. |
| 124 | if (eventHandler != nullptr) { |
| 125 | eventHandler->deleteContext(connectionContext, input_, output_); |
| 126 | } |
| 127 | |
| 128 | // Remove this task from parent bookkeeping |
no test coverage detected