David Reiss | ea2cba8 | 2009-03-30 21:35:00 +0000 | [diff] [blame] | 1 | /* |
| 2 | * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | * or more contributor license agreements. See the NOTICE file |
| 4 | * distributed with this work for additional information |
| 5 | * regarding copyright ownership. The ASF licenses this file |
| 6 | * to you under the Apache License, Version 2.0 (the |
| 7 | * "License"); you may not use this file except in compliance |
| 8 | * with the License. You may obtain a copy of the License at |
| 9 | * |
| 10 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | * |
| 12 | * Unless required by applicable law or agreed to in writing, |
| 13 | * software distributed under the License is distributed on an |
| 14 | * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | * KIND, either express or implied. See the License for the |
| 16 | * specific language governing permissions and limitations |
| 17 | * under the License. |
| 18 | */ |
Mark Slee | 9f0c651 | 2007-02-28 23:58:26 +0000 | [diff] [blame] | 19 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 20 | #include "server/TThreadedServer.h" |
| 21 | #include "transport/TTransportException.h" |
| 22 | #include "concurrency/PosixThreadFactory.h" |
| 23 | |
| 24 | #include <string> |
| 25 | #include <iostream> |
| 26 | #include <pthread.h> |
| 27 | #include <unistd.h> |
| 28 | |
T Jake Luciani | b5e6221 | 2009-01-31 22:36:20 +0000 | [diff] [blame] | 29 | namespace apache { namespace thrift { namespace server { |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 30 | |
Mark Slee | 5ea15f9 | 2007-03-05 22:55:59 +0000 | [diff] [blame] | 31 | using boost::shared_ptr; |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 32 | using namespace std; |
T Jake Luciani | b5e6221 | 2009-01-31 22:36:20 +0000 | [diff] [blame] | 33 | using namespace apache::thrift; |
| 34 | using namespace apache::thrift::protocol; |
| 35 | using namespace apache::thrift::transport; |
| 36 | using namespace apache::thrift::concurrency; |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 37 | |
| 38 | class TThreadedServer::Task: public Runnable { |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 39 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 40 | public: |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 41 | |
| 42 | Task(TThreadedServer& server, |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 43 | shared_ptr<TProcessor> processor, |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 44 | shared_ptr<TProtocol> input, |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 45 | shared_ptr<TProtocol> output, |
| 46 | shared_ptr<TTransport> transport) : |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 47 | server_(server), |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 48 | processor_(processor), |
| 49 | input_(input), |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 50 | output_(output), |
| 51 | transport_(transport) { |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 52 | } |
| 53 | |
| 54 | ~Task() {} |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 55 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 56 | void run() { |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 57 | boost::shared_ptr<TServerEventHandler> eventHandler = |
| 58 | server_.getEventHandler(); |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 59 | void* connectionContext = NULL; |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 60 | if (eventHandler != NULL) { |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 61 | connectionContext = eventHandler->createContext(input_, output_); |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 62 | } |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 63 | try { |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 64 | for (;;) { |
| 65 | if (eventHandler != NULL) { |
| 66 | eventHandler->processContext(connectionContext, transport_); |
| 67 | } |
| 68 | if (!processor_->process(input_, output_, connectionContext) || |
| 69 | !input_->getTransport()->peek()) { |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 70 | break; |
| 71 | } |
| 72 | } |
| 73 | } catch (TTransportException& ttx) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 74 | string errStr = string("TThreadedServer client died: ") + ttx.what(); |
| 75 | GlobalOutput(errStr.c_str()); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 76 | } catch (TException& x) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 77 | string errStr = string("TThreadedServer exception: ") + x.what(); |
| 78 | GlobalOutput(errStr.c_str()); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 79 | } catch (...) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 80 | GlobalOutput("TThreadedServer uncaught exception."); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 81 | } |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 82 | if (eventHandler != NULL) { |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 83 | eventHandler->deleteContext(connectionContext, input_, output_); |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 84 | } |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 85 | |
| 86 | try { |
| 87 | input_->getTransport()->close(); |
| 88 | } catch (TTransportException& ttx) { |
| 89 | string errStr = string("TThreadedServer input close failed: ") + ttx.what(); |
| 90 | GlobalOutput(errStr.c_str()); |
| 91 | } |
| 92 | try { |
| 93 | output_->getTransport()->close(); |
| 94 | } catch (TTransportException& ttx) { |
| 95 | string errStr = string("TThreadedServer output close failed: ") + ttx.what(); |
| 96 | GlobalOutput(errStr.c_str()); |
| 97 | } |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 98 | |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 99 | // Remove this task from parent bookkeeping |
| 100 | { |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 101 | Synchronized s(server_.tasksMonitor_); |
| 102 | server_.tasks_.erase(this); |
| 103 | if (server_.tasks_.empty()) { |
| 104 | server_.tasksMonitor_.notify(); |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 105 | } |
| 106 | } |
| 107 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 108 | } |
| 109 | |
| 110 | private: |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 111 | TThreadedServer& server_; |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 112 | friend class TThreadedServer; |
| 113 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 114 | shared_ptr<TProcessor> processor_; |
| 115 | shared_ptr<TProtocol> input_; |
| 116 | shared_ptr<TProtocol> output_; |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 117 | shared_ptr<TTransport> transport_; |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 118 | }; |
| 119 | |
| 120 | |
| 121 | TThreadedServer::TThreadedServer(shared_ptr<TProcessor> processor, |
| 122 | shared_ptr<TServerTransport> serverTransport, |
| 123 | shared_ptr<TTransportFactory> transportFactory, |
| 124 | shared_ptr<TProtocolFactory> protocolFactory): |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 125 | TServer(processor, serverTransport, transportFactory, protocolFactory), |
| 126 | stop_(false) { |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 127 | threadFactory_ = shared_ptr<PosixThreadFactory>(new PosixThreadFactory()); |
| 128 | } |
| 129 | |
David Reiss | 45d5696 | 2009-03-14 23:35:16 +0000 | [diff] [blame] | 130 | TThreadedServer::TThreadedServer(boost::shared_ptr<TProcessor> processor, |
| 131 | boost::shared_ptr<TServerTransport> serverTransport, |
| 132 | boost::shared_ptr<TTransportFactory> transportFactory, |
| 133 | boost::shared_ptr<TProtocolFactory> protocolFactory, |
| 134 | boost::shared_ptr<ThreadFactory> threadFactory): |
| 135 | TServer(processor, serverTransport, transportFactory, protocolFactory), |
| 136 | threadFactory_(threadFactory), |
| 137 | stop_(false) { |
| 138 | } |
| 139 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 140 | TThreadedServer::~TThreadedServer() {} |
| 141 | |
| 142 | void TThreadedServer::serve() { |
| 143 | |
| 144 | shared_ptr<TTransport> client; |
| 145 | shared_ptr<TTransport> inputTransport; |
| 146 | shared_ptr<TTransport> outputTransport; |
| 147 | shared_ptr<TProtocol> inputProtocol; |
| 148 | shared_ptr<TProtocol> outputProtocol; |
| 149 | |
| 150 | try { |
| 151 | // Start the server listening |
| 152 | serverTransport_->listen(); |
| 153 | } catch (TTransportException& ttx) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 154 | string errStr = string("TThreadedServer::run() listen(): ") +ttx.what(); |
| 155 | GlobalOutput(errStr.c_str()); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 156 | return; |
| 157 | } |
| 158 | |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 159 | // Run the preServe event |
| 160 | if (eventHandler_ != NULL) { |
| 161 | eventHandler_->preServe(); |
| 162 | } |
| 163 | |
| 164 | while (!stop_) { |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 165 | try { |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 166 | client.reset(); |
| 167 | inputTransport.reset(); |
| 168 | outputTransport.reset(); |
| 169 | inputProtocol.reset(); |
| 170 | outputProtocol.reset(); |
| 171 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 172 | // Fetch client from server |
| 173 | client = serverTransport_->accept(); |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 174 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 175 | // Make IO transports |
| 176 | inputTransport = inputTransportFactory_->getTransport(client); |
| 177 | outputTransport = outputTransportFactory_->getTransport(client); |
| 178 | inputProtocol = inputProtocolFactory_->getProtocol(inputTransport); |
| 179 | outputProtocol = outputProtocolFactory_->getProtocol(outputTransport); |
| 180 | |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 181 | TThreadedServer::Task* task = new TThreadedServer::Task(*this, |
| 182 | processor_, |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 183 | inputProtocol, |
David Reiss | 2324871 | 2010-10-06 17:10:08 +0000 | [diff] [blame] | 184 | outputProtocol, |
| 185 | client); |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 186 | |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 187 | // Create a task |
| 188 | shared_ptr<Runnable> runnable = |
| 189 | shared_ptr<Runnable>(task); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 190 | |
| 191 | // Create a thread for this task |
| 192 | shared_ptr<Thread> thread = |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 193 | shared_ptr<Thread>(threadFactory_->newThread(runnable)); |
Mark Slee | b4d3e7b | 2007-11-28 01:51:43 +0000 | [diff] [blame] | 194 | |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 195 | // Insert thread into the set of threads |
| 196 | { |
| 197 | Synchronized s(tasksMonitor_); |
| 198 | tasks_.insert(task); |
| 199 | } |
| 200 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 201 | // Start the thread! |
| 202 | thread->start(); |
| 203 | |
| 204 | } catch (TTransportException& ttx) { |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 205 | if (inputTransport != NULL) { inputTransport->close(); } |
| 206 | if (outputTransport != NULL) { outputTransport->close(); } |
| 207 | if (client != NULL) { client->close(); } |
| 208 | if (!stop_ || ttx.getType() != TTransportException::INTERRUPTED) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 209 | string errStr = string("TThreadedServer: TServerTransport died on accept: ") + ttx.what(); |
| 210 | GlobalOutput(errStr.c_str()); |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 211 | } |
Mark Slee | 907e3d6 | 2007-02-08 22:29:24 +0000 | [diff] [blame] | 212 | continue; |
| 213 | } catch (TException& tx) { |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 214 | if (inputTransport != NULL) { inputTransport->close(); } |
| 215 | if (outputTransport != NULL) { outputTransport->close(); } |
| 216 | if (client != NULL) { client->close(); } |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 217 | string errStr = string("TThreadedServer: Caught TException: ") + tx.what(); |
| 218 | GlobalOutput(errStr.c_str()); |
Mark Slee | 907e3d6 | 2007-02-08 22:29:24 +0000 | [diff] [blame] | 219 | continue; |
| 220 | } catch (string s) { |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 221 | if (inputTransport != NULL) { inputTransport->close(); } |
| 222 | if (outputTransport != NULL) { outputTransport->close(); } |
| 223 | if (client != NULL) { client->close(); } |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 224 | string errStr = "TThreadedServer: Unknown exception: " + s; |
| 225 | GlobalOutput(errStr.c_str()); |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 226 | break; |
| 227 | } |
| 228 | } |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 229 | |
| 230 | // If stopped manually, make sure to close server transport |
| 231 | if (stop_) { |
| 232 | try { |
| 233 | serverTransport_->close(); |
| 234 | } catch (TException &tx) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 235 | string errStr = string("TThreadedServer: Exception shutting down: ") + tx.what(); |
| 236 | GlobalOutput(errStr.c_str()); |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 237 | } |
| 238 | try { |
| 239 | Synchronized s(tasksMonitor_); |
| 240 | while (!tasks_.empty()) { |
| 241 | tasksMonitor_.wait(); |
| 242 | } |
| 243 | } catch (TException &tx) { |
Mark Slee | 2e8a8d4 | 2008-01-16 00:38:20 +0000 | [diff] [blame] | 244 | string errStr = string("TThreadedServer: Exception joining workers: ") + tx.what(); |
| 245 | GlobalOutput(errStr.c_str()); |
Mark Slee | 1d4ce80 | 2007-03-07 05:16:16 +0000 | [diff] [blame] | 246 | } |
| 247 | stop_ = false; |
| 248 | } |
| 249 | |
Mark Slee | b3cb629 | 2007-02-01 22:55:00 +0000 | [diff] [blame] | 250 | } |
| 251 | |
T Jake Luciani | b5e6221 | 2009-01-31 22:36:20 +0000 | [diff] [blame] | 252 | }}} // apache::thrift::server |